diff --git a/examples/foundational/12c-describe-video-anthropic.py b/examples/foundational/12c-describe-video-anthropic.py index 052ffa5d4..cc1f14c92 100644 --- a/examples/foundational/12c-describe-video-anthropic.py +++ b/examples/foundational/12c-describe-video-anthropic.py @@ -16,7 +16,7 @@ from pipecat.pipeline.task import PipelineTask from pipecat.processors.aggregators.user_response import UserResponseAggregator from pipecat.processors.aggregators.vision_image_frame import VisionImageFrameAggregator from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.services.elevenlabs import ElevenLabsTTSService +from pipecat.services.cartesia import CartesiaTTSService from pipecat.services.anthropic import AnthropicLLMService from pipecat.transports.services.daily import DailyParams, DailyTransport from pipecat.vad.silero import SileroVADAnalyzer @@ -72,14 +72,13 @@ async def main(): vision_aggregator = VisionImageFrameAggregator() anthropic = AnthropicLLMService( - api_key=os.getenv("ANTHROPIC_API_KEY"), - model="claude-3-sonnet-20240229" + api_key=os.getenv("ANTHROPIC_API_KEY") ) - tts = ElevenLabsTTSService( - aiohttp_session=session, - api_key=os.getenv("ELEVENLABS_API_KEY"), - voice_id=os.getenv("ELEVENLABS_VOICE_ID"), + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + sample_rate=16000, ) @transport.event_handler("on_first_participant_joined") diff --git a/examples/foundational/14-function-calling.py b/examples/foundational/14-function-calling.py index ea6d57b78..723997e7e 100644 --- a/examples/foundational/14-function-calling.py +++ b/examples/foundational/14-function-calling.py @@ -36,11 +36,11 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def start_fetch_weather(llm): - await llm.push_frame(TextFrame("Let me think.")) +async def start_fetch_weather(llm, function_name): + await llm.push_frame(TextFrame("Let me check on that.")) -async def fetch_weather_from_api(llm, args): +async def fetch_weather_from_api(llm, function_name, args): return {"conditions": "nice", "temperature": "75"} @@ -69,8 +69,11 @@ 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 + # sent to the same callback with an additional function_name parameter. llm.register_function( - "get_current_weather", + #"get_current_weather", + None, fetch_weather_from_api, start_callback=start_fetch_weather) diff --git a/examples/foundational/19a-tools-anthropic.py b/examples/foundational/19a-tools-anthropic.py new file mode 100644 index 000000000..e238de63c --- /dev/null +++ b/examples/foundational/19a-tools-anthropic.py @@ -0,0 +1,122 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import aiohttp +import os +import sys + +from pipecat.frames.frames import LLMMessagesFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.services.cartesia import CartesiaTTSService + +from pipecat.services.anthropic import AnthropicLLMService, AnthropicUserContextAggregator, AnthropicAssistantContextAggregator +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer + +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + + +from runner import configure + +from loguru import logger + +from dotenv import load_dotenv +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def get_weather(function_name, tool_call_id, arguments, context, result_callback): + location = arguments["location"] + await result_callback(f"The weather in {location} is currently 72 degrees and sunny.") + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer() + ) + ) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + sample_rate=16000, + ) + + llm = AnthropicLLMService( + api_key=os.getenv("ANTHROPIC_API_KEY"), + model="claude-3-5-sonnet-20240620" + ) + llm.register_function("get_weather", get_weather) + + tools = [ + { + "name": "get_weather", + "description": "Get the current weather in a given location", + "input_schema": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + } + }, + "required": ["location"], + }, + } + ] + + # todo: test with very short initial user message + + # messages = [{"role": "system", + # "content": "You are a helpful assistant who can report the weather in any location in the universe. Respond concisely. Your response will be turned into speech so use only simple words and punctuation."}, + # {"role": "user", + # "content": " Start the conversation by introducing yourself."}] + + messages = [{"role": "user", "content": "Say 'hello' to start the conversation."}] + + context = OpenAILLMContext(messages, tools) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline([ + transport.input(), # Transport user input + context_aggregator.user(), # User speech to text + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses and tool context + ]) + + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True)) + + @ transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/foundational/19b-tools-video-anthropic.py b/examples/foundational/19b-tools-video-anthropic.py new file mode 100644 index 000000000..26d466e9e --- /dev/null +++ b/examples/foundational/19b-tools-video-anthropic.py @@ -0,0 +1,183 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import aiohttp +import os +import sys + +from pipecat.frames.frames import LLMMessagesFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.services.cartesia import CartesiaTTSService + +from pipecat.services.anthropic import AnthropicLLMService, AnthropicUserContextAggregator, AnthropicAssistantContextAggregator +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer + +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + + +from runner import configure + +from loguru import logger + +from dotenv import load_dotenv +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") +# logger.add(sys.stderr, level="TRACE") + +video_participant_id = None + +# globally declare llm so that we can access it in the get_image function +llm = None + + +async def get_weather(function_name, tool_call_id, arguments, context, result_callback): + location = arguments["location"] + await result_callback(f"The weather in {location} is currently 72 degrees and sunny.") + + +async def get_image(function_name, tool_call_id, arguments, context, result_callback): + global llm + question = arguments["question"] + await llm.request_image_frame(user_id=video_participant_id, text_content=question) + + +async def main(): + global llm + + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer() + ) + ) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + sample_rate=16000, + ) + + llm = AnthropicLLMService( + api_key=os.getenv("ANTHROPIC_API_KEY"), + model="claude-3-5-sonnet-20240620", + enable_prompt_caching_beta=True + ) + llm.register_function("get_weather", get_weather) + llm.register_function("get_image", get_image) + + tools = [ + { + "name": "get_weather", + "description": "Get the current weather in a given location", + "input_schema": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + } + }, + "required": ["location"], + }, + }, + { + "name": "get_image", + "description": "Get an image from the video stream.", + "input_schema": { + "type": "object", + "properties": { + "question": { + "type": "string", + "description": "The question that the user is asking about the image.", + } + }, + "required": ["question"], + }, + } + ] + + # todo: test with very short initial user message + + system_prompt = """\ +You are a helpful assistant who converses with a user and answers questions. Respond concisely to general questions. + +Your response will be turned into speech so use only simple words and punctuation. + +You have access to two tools: get_weather and get_image. + +You can respond to questions about the weather using the get_weather tool. + +You can answer questions about the user's video stream using the get_image tool. Some examples of phrases that \ +indicate you should use the get_image tool are: + - What do you see? + - What's in the video? + - Can you describe the video? + - Tell me about what you see. + - Tell me something interesting about what you see. + - What's happening in the video? + +If you need to use a tool, simply use the tool. Do not tell the user the tool you are using. Be brief and concise. + """ + + messages = [ + { + "role": "system", + "content": [ + { + "type": "text", + "text": system_prompt, + } + ] + }, + { + "role": "user", + "content": "Start the conversation by introducing yourself." + }] + + context = OpenAILLMContext(messages, tools) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline([ + transport.input(), # Transport user input + context_aggregator.user(), # User speech to text + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses and tool context + ]) + + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True)) + + @ transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + global video_participant_id + video_participant_id = participant["id"] + transport.capture_participant_transcription(video_participant_id) + transport.capture_participant_video(video_participant_id, framerate=0) + # Kick off the conversation. + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + runner = PipelineRunner() + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/foundational/19c-tools-togetherai.py b/examples/foundational/19c-tools-togetherai.py new file mode 100644 index 000000000..c1ef328b9 --- /dev/null +++ b/examples/foundational/19c-tools-togetherai.py @@ -0,0 +1,137 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import aiohttp +import os +import sys +import json + +from pipecat.frames.frames import LLMMessagesFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.services.cartesia import CartesiaTTSService + +from pipecat.services.together import TogetherLLMService, TogetherContextAggregatorPair +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer + +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + + +from runner import configure + +from loguru import logger + +from dotenv import load_dotenv +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def get_current_weather(function_name, tool_call_id, arguments, context, result_callback): + logger.debug("IN get_current_weather") + location = arguments["location"] + await result_callback(f"The weather in {location} is currently 72 degrees and sunny.") + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer() + ) + ) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + sample_rate=16000, + ) + + llm = TogetherLLMService( + api_key=os.getenv("TOGETHER_API_KEY"), + model=os.getenv("TOGETHER_MODEL"), + ) + llm.register_function("get_current_weather", get_current_weather) + + weatherTool = { + "name": "get_current_weather", + "description": "Get the current weather in a given location", + "parameters": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + "required": ["location"], + }, + } + + system_prompt = f"""\ +You have access to the following functions: + +Use the function '{weatherTool["name"]}' to '{weatherTool["description"]}': +{json.dumps(weatherTool)} + +If you choose to call a function ONLY reply in the following format with no prefix or suffix: + +{{\"example_name\": \"example_value\"}} + +Reminder: +- Function calls MUST follow the specified format, start with +- Required parameters MUST be specified +- Only call one function at a time +- Put the entire function call reply on one line +- If there is no function call available, answer the question like normal with your current knowledge and do not tell the user about function calls + +""" + + messages = [{"role": "system", + "content": system_prompt}, + {"role": "user", + "content": "Wait for the user to say something."}] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline([ + transport.input(), # Transport user input + context_aggregator.user(), # User speech to text + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses and tool context + ]) + + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True)) + + @ transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + await task.queue_frames([LLMMessagesFrame(messages)]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/linux-py3.10-requirements.txt b/linux-py3.10-requirements.txt index 7b42fd5e3..dd9969964 100644 --- a/linux-py3.10-requirements.txt +++ b/linux-py3.10-requirements.txt @@ -16,7 +16,7 @@ aiosignal==1.3.1 # via aiohttp annotated-types==0.7.0 # via pydantic -anthropic==0.28.1 +anthropic==0.34.0 # via # openpipe # pipecat-ai (pyproject.toml) diff --git a/macos-py3.10-requirements.txt b/macos-py3.10-requirements.txt index a764e62d4..c94f6e41a 100644 --- a/macos-py3.10-requirements.txt +++ b/macos-py3.10-requirements.txt @@ -16,7 +16,7 @@ aiosignal==1.3.1 # via aiohttp annotated-types==0.7.0 # via pydantic -anthropic==0.28.1 +anthropic==0.34.0 # via # openpipe # pipecat-ai (pyproject.toml) diff --git a/pyproject.toml b/pyproject.toml index 8b9c6cb64..31ca9294e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,7 +34,7 @@ Source = "https://github.com/pipecat-ai/pipecat" Website = "https://pipecat.ai" [project.optional-dependencies] -anthropic = [ "anthropic~=0.28.1" ] +anthropic = [ "anthropic~=0.34.0" ] azure = [ "azure-cognitiveservices-speech~=1.38.0" ] cartesia = [ "websockets~=12.0" ] daily = [ "daily-python~=0.10.1" ] @@ -51,6 +51,7 @@ openai = [ "openai~=1.35.0" ] openpipe = [ "openpipe~=4.18.0" ] playht = [ "pyht~=0.0.28" ] silero = [ "silero-vad~=5.1" ] +together = [ "together~=1.2.7" ] websocket = [ "websockets~=12.0", "fastapi~=0.111.0" ] whisper = [ "faster-whisper~=1.0.3" ] xtts = [ "resampy~=0.4.3" ] diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 90127492b..68ec1ec38 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -4,7 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # -from typing import Any, List, Mapping, Tuple +from typing import Any, List, Mapping, Tuple, Optional from dataclasses import dataclass, field @@ -177,6 +177,22 @@ class LLMMessagesUpdateFrame(DataFrame): messages: List[dict] +@dataclass +class LLMSetToolsFrame(DataFrame): + """A frame containing a list of tools for an LLM to use for function calling. + The specific format depends on the LLM being used, but it should typically + contain JSON Schema objects. + """ + tools: List[dict] + + +@dataclass +class LLMEnablePromptCachingFrame(DataFrame): + """A frame to enable/disable prompt caching in certain LLMs. + """ + enable: bool + + @dataclass class TTSSpeakFrame(DataFrame): """A frame that contains a text that should be spoken by the TTS in the @@ -189,6 +205,7 @@ class TTSSpeakFrame(DataFrame): @dataclass class TransportMessageFrame(DataFrame): message: Any + urgent: bool = False def __str__(self): return f"{self.name}(message: {self.message})" @@ -222,7 +239,7 @@ class CancelFrame(SystemFrame): class ErrorFrame(SystemFrame): """This is used notify upstream that an error has occurred downstream the pipeline.""" - error: str | None + error: str def __str__(self): return f"{self.name}(error: {self.error})" @@ -230,9 +247,9 @@ class ErrorFrame(SystemFrame): @dataclass class StopTaskFrame(SystemFrame): - """Indicates that a pipeline task should be stopped. This should inform the - pipeline processors that they should stop pushing frames but that they - should be kept in a running state. + """Indicates that a pipeline task should be stopped but that the pipeline + processors should be kept in a running state. This is normally queued from + the pipeline task. """ pass @@ -389,6 +406,7 @@ class TTSStoppedFrame(ControlFrame): class UserImageRequestFrame(ControlFrame): """A frame user to request an image from the given user.""" user_id: str + context: Optional[any] def __str__(self): return f"{self.name}, user: {self.user_id}" @@ -406,3 +424,22 @@ class TTSVoiceUpdateFrame(ControlFrame): """A control frame containing a request to update to a new TTS voice. """ voice: str + + +@dataclass +class FunctionCallInProgressFrame(SystemFrame): + """A frame signaling that a function call is in progress. + """ + function_name: str + tool_call_id: str + arguments: str + + +@dataclass +class FunctionCallResultFrame(DataFrame): + """A frame containing the result of an LLM function (tool) call. + """ + function_name: str + tool_call_id: str + arguments: str + result: any diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index cdd3eafb2..1f09ad233 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -10,7 +10,14 @@ from typing import AsyncIterable, Iterable from pydantic import BaseModel -from pipecat.frames.frames import CancelFrame, EndFrame, ErrorFrame, Frame, MetricsFrame, StartFrame, StopTaskFrame +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + ErrorFrame, + Frame, + MetricsFrame, + StartFrame, + StopTaskFrame) from pipecat.pipeline.base_pipeline import BasePipeline from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.utils.utils import obj_count, obj_id @@ -37,10 +44,18 @@ class Source(FrameProcessor): match direction: case FrameDirection.UPSTREAM: - await self._up_queue.put(frame) + await self._handle_upstream_frame(frame) case FrameDirection.DOWNSTREAM: await self.push_frame(frame, direction) + async def _handle_upstream_frame(self, frame: Frame): + if isinstance(frame, ErrorFrame): + logger.error(f"Error running app: {frame.error}") + # Cancel all tasks downstream. + await self.push_frame(CancelFrame()) + # Tell the task we should stop. + await self._up_queue.put(StopTaskFrame()) + class PipelineTask: @@ -70,7 +85,7 @@ class PipelineTask: # Make sure everything is cleaned up downstream. This is sent # out-of-band from the main streaming task which is what we want since # we want to cancel right away. - await self._source.process_frame(CancelFrame(), FrameDirection.DOWNSTREAM) + await self._source.push_frame(CancelFrame()) self._process_down_task.cancel() self._process_up_task.cancel() await self._process_down_task @@ -92,8 +107,6 @@ class PipelineTask: elif isinstance(frames, Iterable): for frame in frames: await self.queue_frame(frame) - else: - raise Exception("Frames must be an iterable or async iterable") def _initial_metrics_frame(self) -> MetricsFrame: processors = self._pipeline.processors_with_metrics() @@ -110,7 +123,7 @@ class PipelineTask: ) await self._source.process_frame(start_frame, FrameDirection.DOWNSTREAM) - if self._params.send_initial_empty_metrics: + if self._params.enable_metrics and self._params.send_initial_empty_metrics: await self._source.process_frame(self._initial_metrics_frame(), FrameDirection.DOWNSTREAM) running = True @@ -136,9 +149,8 @@ class PipelineTask: while True: try: frame = await self._up_queue.get() - if isinstance(frame, ErrorFrame): - logger.error(f"Error running app: {frame.error}") - await self.queue_frame(CancelFrame()) + if isinstance(frame, StopTaskFrame): + await self.queue_frame(StopTaskFrame()) self._up_queue.task_done() except asyncio.CancelledError: break diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 6939a70c4..7c38e62ad 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -4,9 +4,10 @@ # SPDX-License-Identifier: BSD 2-Clause License # +import sys from typing import List -from pipecat.services.openai import OpenAILLMContextFrame, OpenAILLMContext +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame, OpenAILLMContext from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.frames.frames import ( @@ -17,6 +18,7 @@ from pipecat.frames.frames import ( LLMMessagesAppendFrame, LLMMessagesFrame, LLMMessagesUpdateFrame, + LLMSetToolsFrame, StartInterruptionFrame, TranscriptionFrame, TextFrame, @@ -123,18 +125,11 @@ class LLMResponseAggregator(FrameProcessor): self._reset() await self.push_frame(frame, direction) elif isinstance(frame, LLMMessagesAppendFrame): - self._messages.extend(frame.messages) - messages_frame = LLMMessagesFrame(self._messages) - await self.push_frame(messages_frame) + self._add_messages(frame.messages) elif isinstance(frame, LLMMessagesUpdateFrame): - # We push the frame downstream so the assistant aggregator gets - # updated as well. - await self.push_frame(frame) - # We can now reset this one. - self._reset() - self._messages = frame.messages - messages_frame = LLMMessagesFrame(self._messages) - await self.push_frame(messages_frame) + self._set_messages(frame.messages) + elif isinstance(frame, LLMSetToolsFrame): + self._set_tools(frame.tools) else: await self.push_frame(frame, direction) @@ -152,6 +147,19 @@ class LLMResponseAggregator(FrameProcessor): frame = LLMMessagesFrame(self._messages) await self.push_frame(frame) + # TODO-CB: Types + 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): + # noop in the base class + pass + def _reset(self): self._aggregation = "" self._aggregating = False @@ -240,9 +248,29 @@ class LLMFullResponseAggregator(FrameProcessor): class LLMContextAggregator(LLMResponseAggregator): def __init__(self, *, context: OpenAILLMContext, **kwargs): - - self._context = context super().__init__(**kwargs) + self._context = context + + @property + def context(self): + return self._context + + def get_context_frame(self) -> OpenAILLMContextFrame: + return OpenAILLMContextFrame(context=self._context) + + async def push_context_frame(self): + frame = self.get_context_frame() + await self.push_frame(frame) + + # TODO-CB: Types + def _add_messages(self, messages): + self._context.add_messages(messages) + + def _set_messages(self, messages): + self._context.set_messages(messages) + + def _set_tools(self, tools: List): + self._context.set_tools(tools) async def _push_aggregation(self): if len(self._aggregation) > 0: diff --git a/src/pipecat/processors/aggregators/openai_llm_context.py b/src/pipecat/processors/aggregators/openai_llm_context.py index 65c8da6ad..009040996 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/src/pipecat/processors/aggregators/openai_llm_context.py @@ -12,7 +12,9 @@ from typing import List from PIL import Image -from pipecat.frames.frames import Frame, VisionImageRawFrame +from pipecat.frames.frames import Frame, VisionImageRawFrame, FunctionCallInProgressFrame, FunctionCallResultFrame +from pipecat.processors.frame_processor import FrameProcessor + from openai._types import NOT_GIVEN, NotGiven @@ -42,20 +44,19 @@ class OpenAILLMContext: tools: List[ChatCompletionToolParam] | NotGiven = NOT_GIVEN, tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = NOT_GIVEN ): - self.messages: List[ChatCompletionMessageParam] = messages if messages else [ + self._messages: List[ChatCompletionMessageParam] = messages if messages else [ ] - self.tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = tool_choice - self.tools: List[ChatCompletionToolParam] | NotGiven = tools + self._tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = tool_choice + self._tools: List[ChatCompletionToolParam] | NotGiven = tools @staticmethod def from_messages(messages: List[dict]) -> "OpenAILLMContext": context = OpenAILLMContext() + for message in messages: - context.add_message({ - "content": message["content"], - "role": message["role"], - "name": message["name"] if "name" in message else message["role"] - }) + if "name" not in message: + message["name"] = message["role"] + context.add_message(message) return context @staticmethod @@ -83,25 +84,70 @@ class OpenAILLMContext: }) return context + @property + def messages(self) -> List[ChatCompletionMessageParam]: + return self._messages + + @property + def tools(self) -> List[ChatCompletionToolParam] | NotGiven: + return self._tools + + @property + def tool_choice(self) -> ChatCompletionToolChoiceOptionParam | NotGiven: + return self._tool_choice + def add_message(self, message: ChatCompletionMessageParam): - self.messages.append(message) + self._messages.append(message) + + def add_messages(self, messages: List[ChatCompletionMessageParam]): + self._messages.extend(messages) + + def set_messages(self, messages: List[ChatCompletionMessageParam]): + self._messages[:] = messages def get_messages(self) -> List[ChatCompletionMessageParam]: - return self.messages + return self._messages def get_messages_json(self) -> str: - return json.dumps(self.messages, cls=CustomEncoder) + return json.dumps(self._messages, cls=CustomEncoder) def set_tool_choice( self, tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven ): - self.tool_choice = tool_choice + self._tool_choice = tool_choice def set_tools(self, tools: List[ChatCompletionToolParam] | NotGiven = NOT_GIVEN): if tools != NOT_GIVEN and len(tools) == 0: tools = NOT_GIVEN + self._tools = tools - self.tools = tools + async def call_function( + self, + f: callable, + *, + function_name: str, + tool_call_id: str, + arguments: str, + llm: FrameProcessor) -> None: + + # 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). + await llm.push_frame(FunctionCallInProgressFrame( + function_name=function_name, + tool_call_id=tool_call_id, + arguments=arguments, + )) + + # Define a callback function that pushes a FunctionCallResultFrame downstream. + async def function_call_result_callback(result): + await llm.push_frame(FunctionCallResultFrame( + function_name=function_name, + tool_call_id=tool_call_id, + arguments=arguments, + result=result)) + await f(function_name=function_name, tool_call_id=tool_call_id, arguments=arguments, + context=self, result_callback=function_call_result_callback) @dataclass diff --git a/src/pipecat/processors/aggregators/sentence.py b/src/pipecat/processors/aggregators/sentence.py index a7992eace..7ee641826 100644 --- a/src/pipecat/processors/aggregators/sentence.py +++ b/src/pipecat/processors/aggregators/sentence.py @@ -4,10 +4,9 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import re - from pipecat.frames.frames import EndFrame, Frame, InterimTranscriptionFrame, TextFrame from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.utils.string import match_endofsentence class SentenceAggregator(FrameProcessor): @@ -40,12 +39,10 @@ class SentenceAggregator(FrameProcessor): return if isinstance(frame, TextFrame): - m = re.search("(.*[?.!])(.*)", frame.text) - if m: - await self.push_frame(TextFrame(self._aggregation + m.group(1))) - self._aggregation = m.group(2) - else: - self._aggregation += frame.text + self._aggregation += frame.text + if match_endofsentence(self._aggregation): + await self.push_frame(TextFrame(self._aggregation)) + self._aggregation = "" elif isinstance(frame, EndFrame): if self._aggregation: await self.push_frame(TextFrame(self._aggregation)) diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 2e316f1c4..f482aac6a 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -5,53 +5,46 @@ # import asyncio -import dataclasses -from typing import Any, Awaitable, Callable, Dict, List, Literal, Optional, Type -from pydantic import PrivateAttr, BaseModel, ValidationError +from typing import Any, Awaitable, Callable, Dict, List, Literal, Optional, Union +from pydantic import BaseModel, Field, PrivateAttr, ValidationError from pipecat.frames.frames import ( BotInterruptionFrame, + BotStartedSpeakingFrame, + BotStoppedSpeakingFrame, CancelFrame, EndFrame, + ErrorFrame, Frame, InterimTranscriptionFrame, - LLMFullResponseEndFrame, - LLMFullResponseStartFrame, - LLMMessagesAppendFrame, - LLMMessagesUpdateFrame, - LLMModelUpdateFrame, - MetricsFrame, StartFrame, SystemFrame, - TTSSpeakFrame, - TTSVoiceUpdateFrame, - TextFrame, TranscriptionFrame, TransportMessageFrame, UserStartedSpeakingFrame, + FunctionCallResultFrame, UserStoppedSpeakingFrame) -from pipecat.pipeline.pipeline import Pipeline -from pipecat.processors.aggregators.llm_response import ( - LLMAssistantResponseAggregator, LLMUserResponseAggregator) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.services.cartesia import CartesiaTTSService -from pipecat.services.openai import OpenAILLMService, OpenAILLMContext from pipecat.transports.base_transport import BaseTransport from loguru import logger +RTVI_PROTOCOL_VERSION = "0.1" + +ActionResult = Union[bool, int, float, str, list, dict] + + class RTVIServiceOption(BaseModel): name: str - handler: Optional[Callable[['RTVIProcessor', - 'RTVIServiceOptionConfig'], - Awaitable[None]]] = None + type: Literal["bool", "number", "string", "array", "object"] + handler: Callable[["RTVIProcessor", str, "RTVIServiceOptionConfig"], + Awaitable[None]] = Field(exclude=True) class RTVIService(BaseModel): name: str - cls: Type[FrameProcessor] options: List[RTVIServiceOption] _options_dict: Dict[str, RTVIServiceOption] = PrivateAttr(default={}) @@ -61,6 +54,33 @@ class RTVIService(BaseModel): self._options_dict[option.name] = option return super().model_post_init(__context) + +class RTVIActionArgumentData(BaseModel): + name: str + value: Any + + +class RTVIActionArgument(BaseModel): + name: str + type: Literal["bool", "number", "string", "array", "object"] + + +class RTVIAction(BaseModel): + service: str + action: str + arguments: List[RTVIActionArgument] = [] + result: Literal["bool", "number", "string", "array", "object"] + handler: Callable[["RTVIProcessor", str, Dict[str, Any]], + Awaitable[ActionResult]] = Field(exclude=True) + _arguments_dict: Dict[str, RTVIActionArgument] = PrivateAttr(default={}) + + def model_post_init(self, __context: Any) -> None: + self._arguments_dict = {} + for arg in self.arguments: + self._arguments_dict[arg.name] = arg + return super().model_post_init(__context) + + # # Client -> Pipecat messages. # @@ -78,22 +98,17 @@ class RTVIServiceConfig(BaseModel): class RTVIConfig(BaseModel): config: List[RTVIServiceConfig] - _config_dict: Dict[str, RTVIServiceConfig] = PrivateAttr(default={}) - - def model_post_init(self, __context: Any) -> None: - self._config_dict = {} - for c in self.config: - self._config_dict[c.service] = c - return super().model_post_init(__context) -class RTVILLMContextData(BaseModel): - messages: List[dict] +class RTVIActionRunArgument(BaseModel): + name: str + value: Any -class RTVITTSSpeakData(BaseModel): - text: str - interrupt: Optional[bool] = False +class RTVIActionRun(BaseModel): + service: str + action: str + arguments: Optional[List[RTVIActionRunArgument]] = None class RTVIMessage(BaseModel): @@ -107,16 +122,15 @@ class RTVIMessage(BaseModel): # -class RTVIResponseData(BaseModel): - success: bool +class RTVIErrorResponseData(BaseModel): error: Optional[str] = None -class RTVIResponse(BaseModel): +class RTVIErrorResponse(BaseModel): label: Literal["rtvi-ai"] = "rtvi-ai" - type: Literal["response"] = "response" + type: Literal["error-response"] = "error-response" id: str - data: RTVIResponseData + data: RTVIErrorResponseData class RTVIErrorData(BaseModel): @@ -129,29 +143,84 @@ class RTVIError(BaseModel): data: RTVIErrorData -class RTVILLMContextMessageData(BaseModel): - messages: List[dict] +class RTVIDescribeConfigData(BaseModel): + config: List[RTVIService] -class RTVILLMContextMessage(BaseModel): +class RTVIDescribeConfig(BaseModel): label: Literal["rtvi-ai"] = "rtvi-ai" - type: Literal["llm-context"] = "llm-context" - data: RTVILLMContextMessageData + type: Literal["config-available"] = "config-available" + id: str + data: RTVIDescribeConfigData -class RTVITTSTextMessageData(BaseModel): - text: str +class RTVIDescribeActionsData(BaseModel): + actions: List[RTVIAction] -class RTVITTSTextMessage(BaseModel): +class RTVIDescribeActions(BaseModel): label: Literal["rtvi-ai"] = "rtvi-ai" - type: Literal["tts-text"] = "tts-text" - data: RTVITTSTextMessageData + type: Literal["actions-available"] = "actions-available" + id: str + data: RTVIDescribeActionsData + + +class RTVIConfigResponse(BaseModel): + label: Literal["rtvi-ai"] = "rtvi-ai" + type: Literal["config"] = "config" + id: str + data: RTVIConfig + + +class RTVIActionResponseData(BaseModel): + result: ActionResult + + +class RTVIActionResponse(BaseModel): + label: Literal["rtvi-ai"] = "rtvi-ai" + type: Literal["action-response"] = "action-response" + id: str + data: RTVIActionResponseData + + +class RTVIBotReadyData(BaseModel): + version: str + config: List[RTVIServiceConfig] class RTVIBotReady(BaseModel): label: Literal["rtvi-ai"] = "rtvi-ai" type: Literal["bot-ready"] = "bot-ready" + data: RTVIBotReadyData + + +class RTVILLMFunctionCallMessageData(BaseModel): + function_name: str + tool_call_id: str + args: dict + + +class RTVILLMFunctionCallMessage(BaseModel): + label: Literal["rtvi-ai"] = "rtvi-ai" + type: Literal["llm-function-call"] = "llm-function-call" + data: RTVILLMFunctionCallMessageData + + +class RTVILLMFunctionCallStartMessageData(BaseModel): + function_name: str + + +class RTVILLMFunctionCallStartMessage(BaseModel): + label: Literal["rtvi-ai"] = "rtvi-ai" + type: Literal["llm-function-call-start"] = "llm-function-call-start" + data: RTVILLMFunctionCallStartMessageData + + +class RTVILLMFunctionCallResultData(BaseModel): + function_name: str + tool_call_id: str + arguments: dict + result: dict class RTVITranscriptionMessageData(BaseModel): @@ -177,177 +246,86 @@ class RTVIUserStoppedSpeakingMessage(BaseModel): type: Literal["user-stopped-speaking"] = "user-stopped-speaking" -class RTVIJSONCompletion(BaseModel): +class RTVIBotStartedSpeakingMessage(BaseModel): label: Literal["rtvi-ai"] = "rtvi-ai" - type: Literal["json-completion"] = "json-completion" - data: str + type: Literal["bot-started-speaking"] = "bot-started-speaking" -class FunctionCaller(FrameProcessor): - - def __init__(self, context): - super().__init__() - self._checking = False - self._aggregating = False - self._emitted_start = False - self._aggregation = "" - self._context = context - - self._callbacks = {} - self._start_callbacks = {} - - def register_function(self, function_name: str, callback, start_callback=None): - self._callbacks[function_name] = callback - if start_callback: - self._start_callbacks[function_name] = start_callback - - def unregister_function(self, function_name: str): - del self._callbacks[function_name] - if self._start_callbacks[function_name]: - del self._start_callbacks[function_name] - - def has_function(self, function_name: str): - return function_name in self._callbacks.keys() - - async def call_function(self, function_name: str, args): - if function_name in self._callbacks.keys(): - return await self._callbacks[function_name](self, args) - return None - - async def call_start_function(self, function_name: str): - if function_name in self._start_callbacks.keys(): - await self._start_callbacks[function_name](self) - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if isinstance(frame, LLMFullResponseStartFrame): - self._checking = True - await self.push_frame(frame, direction) - elif isinstance(frame, TextFrame) and self._checking: - # TODO-CB: should we expand this to any non-text character to start the completion? - if frame.text.strip().startswith("{") or frame.text.strip().startswith("```"): - self._emitted_start = False - self._checking = False - self._aggregation = frame.text - self._aggregating = True - else: - self._checking = False - self._aggregating = False - self._aggregation = "" - self._emitted_start = False - await self.push_frame(frame, direction) - elif isinstance(frame, TextFrame) and self._aggregating: - self._aggregation += frame.text - # TODO-CB: We can probably ignore function start I think - # if not self._emitted_start: - # fn = re.search(r'{"function_name":\s*"(.*)",', self._aggregation) - # if fn and fn.group(1): - # await self.call_start_function(fn.group(1)) - # self._emitted_start = True - elif isinstance(frame, LLMFullResponseEndFrame) and self._aggregating: - try: - self._aggregation = self._aggregation.replace("```json", "").replace("```", "") - self._context.add_message({"role": "assistant", "content": self._aggregation}) - message = RTVIJSONCompletion(data=self._aggregation) - msg = message.model_dump(exclude_none=True) - await self.push_frame(TransportMessageFrame(message=msg)) - - except Exception as e: - print(f"Error parsing function call json: {e}") - print(f"aggregation was: {self._aggregation}") - - self._aggregating = False - self._aggregation = "" - self._emitted_start = False - elif isinstance(frame, LLMFullResponseEndFrame): - await self.push_frame(frame, direction) - else: - await self.push_frame(frame, direction) +class RTVIBotStoppedSpeakingMessage(BaseModel): + label: Literal["rtvi-ai"] = "rtvi-ai" + type: Literal["bot-stopped-speaking"] = "bot-stopped-speaking" -class RTVITTSTextProcessor(FrameProcessor): - - def __init__(self): - super().__init__() - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - await self.push_frame(frame, direction) - - if isinstance(frame, TextFrame): - message = RTVITTSTextMessage(data=RTVITTSTextMessageData(text=frame.text)) - await self.push_frame(TransportMessageFrame(message=message.model_dump(exclude_none=True))) - - -async def handle_llm_model_update(rtvi: 'RTVIProcessor', option: RTVIServiceOptionConfig): - frame = LLMModelUpdateFrame(option.value) - await rtvi.push_frame(frame) - - -async def handle_llm_messages_update(rtvi: 'RTVIProcessor', option: RTVIServiceOptionConfig): - frame = LLMMessagesUpdateFrame(option.value) - await rtvi.push_frame(frame) - - -async def handle_tts_voice_update(rtvi: 'RTVIProcessor', option: RTVIServiceOptionConfig): - frame = TTSVoiceUpdateFrame(option.value) - await rtvi.push_frame(frame) - -DEFAULT_LLM_SERVICE = RTVIService( - name="llm", - cls=OpenAILLMService, - options=[ - RTVIServiceOption(name="model", handler=handle_llm_model_update), - RTVIServiceOption(name="messages", handler=handle_llm_messages_update) - ]) - -DEFAULT_TTS_SERVICE = RTVIService( - name="tts", - cls=CartesiaTTSService, - options=[ - RTVIServiceOption(name="voice_id", handler=handle_tts_voice_update), - ]) +class RTVIProcessorParams(BaseModel): + send_bot_ready: bool = True class RTVIProcessor(FrameProcessor): - def __init__(self, *, transport: BaseTransport): + def __init__(self, + *, + transport: BaseTransport, + config: RTVIConfig = RTVIConfig(config=[]), + params: RTVIProcessorParams = RTVIProcessorParams()): super().__init__() - self._transport = transport - self._config: RTVIConfig | None = None - self._ctor_args: Dict[str, Any] = {} + self._config = config + self._params = params - self._start_frame: Frame | None = None self._pipeline: FrameProcessor | None = None - self._first_participant_joined: bool = False + self._pipeline_started = False + self._transport_joined = False - # Register transport event so we can send a `bot-ready` event (and maybe - # others) when the participant joins. - transport.add_event_handler( - "on_first_participant_joined", - self._on_first_participant_joined) - - # Register default services. + self._registered_actions: Dict[str, RTVIAction] = {} self._registered_services: Dict[str, RTVIService] = {} - self.register_service(DEFAULT_LLM_SERVICE) - self.register_service(DEFAULT_TTS_SERVICE) - self._frame_handler_task = self.get_event_loop().create_task(self._frame_handler()) - self._frame_queue = asyncio.Queue() + self._push_frame_task = self.get_event_loop().create_task(self._push_frame_task_handler()) + self._push_queue = asyncio.Queue() + + self._message_task = self.get_event_loop().create_task(self._message_task_handler()) + self._message_queue = asyncio.Queue() + + # TODO(aleix): This is very Daily specific. There should be a generic + # way to do this. + transport.add_event_handler("on_joined", self._transport_on_joined) + + def register_action(self, action: RTVIAction): + id = self._action_id(action.service, action.action) + self._registered_actions[id] = action def register_service(self, service: RTVIService): self._registered_services[service.name] = service - def setup_on_start(self, config: RTVIConfig | None, ctor_args: Dict[str, Any]): - self._config = config - self._ctor_args = ctor_args + async def interrupt_bot(self): + await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) - async def update_config(self, config: RTVIConfig): - if self._pipeline: - await self._handle_config_update(config) - self._config = config + async def send_error(self, error: str): + message = RTVIError(data=RTVIErrorData(message=error)) + await self._push_transport_message(message) + + async def handle_function_call( + self, + function_name: str, + tool_call_id: str, + arguments: dict, + context, + result_callback): + fn = RTVILLMFunctionCallMessageData( + function_name=function_name, + tool_call_id=tool_call_id, + args=arguments) + message = RTVILLMFunctionCallMessage(data=fn) + await self._push_transport_message(message, exclude_none=False) + + async def handle_function_call_start(self, function_name: str): + fn = RTVILLMFunctionCallStartMessageData(function_name=function_name) + message = RTVILLMFunctionCallStartMessage(data=fn) + await self._push_transport_message(message, exclude_none=False) + + async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): + if isinstance(frame, SystemFrame): + await super().push_frame(frame, direction) + else: + await self._internal_push_frame(frame, direction) async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -356,71 +334,85 @@ class RTVIProcessor(FrameProcessor): if isinstance(frame, CancelFrame): await self._cancel(frame) await self.push_frame(frame, direction) + elif isinstance(frame, ErrorFrame): + await self.send_error(frame.error) + await self.push_frame(frame, direction) # All other system frames elif isinstance(frame, SystemFrame): await self.push_frame(frame, direction) # Control frames elif isinstance(frame, StartFrame): await self._start(frame) - await self._internal_push_frame(frame, direction) + await self.push_frame(frame, direction) elif isinstance(frame, EndFrame): # Push EndFrame before stop(), because stop() waits on the task to # finish and the task finishes when EndFrame is processed. - await self._internal_push_frame(frame, direction) + await self.push_frame(frame, direction) await self._stop(frame) + elif isinstance(frame, UserStartedSpeakingFrame) or isinstance(frame, UserStoppedSpeakingFrame): + await self._handle_interruptions(frame) + await self.push_frame(frame, direction) + elif isinstance(frame, BotStartedSpeakingFrame) or isinstance(frame, BotStoppedSpeakingFrame): + await self._handle_bot_speaking(frame) + await self.push_frame(frame, direction) + # Data frames + elif isinstance(frame, TranscriptionFrame) or isinstance(frame, InterimTranscriptionFrame): + await self._handle_transcriptions(frame) + await self.push_frame(frame, direction) + elif isinstance(frame, TransportMessageFrame): + await self._message_queue.put(frame) # Other frames else: - await self._internal_push_frame(frame, direction) + await self.push_frame(frame, direction) async def cleanup(self): if self._pipeline: await self._pipeline.cleanup() async def _start(self, frame: StartFrame): - try: - await self._handle_pipeline_setup(frame, self._config) - except Exception as e: - await self._send_error(f"unable to setup RTVI pipeline: {e}") + self._pipeline_started = True + await self._update_config(self._config) + await self._maybe_send_bot_ready() async def _stop(self, frame: EndFrame): - await self._frame_handler_task + # We need to cancel the message task handler because that one is not + # processing EndFrames. + self._message_task.cancel() + await self._message_task + await self._push_frame_task async def _cancel(self, frame: CancelFrame): - self._frame_handler_task.cancel() - await self._frame_handler_task + self._message_task.cancel() + await self._message_task + self._push_frame_task.cancel() + await self._push_frame_task async def _internal_push_frame( self, frame: Frame | None, direction: FrameDirection | None = FrameDirection.DOWNSTREAM): - await self._frame_queue.put((frame, direction)) + await self._push_queue.put((frame, direction)) - async def _frame_handler(self): + async def _push_frame_task_handler(self): running = True while running: try: - (frame, direction) = await self._frame_queue.get() - await self._handle_frame(frame, direction) - self._frame_queue.task_done() + (frame, direction) = await self._push_queue.get() + await super().push_frame(frame, direction) + self._push_queue.task_done() running = not isinstance(frame, EndFrame) except asyncio.CancelledError: break - async def _handle_frame(self, frame: Frame, direction: FrameDirection): - if isinstance(frame, TransportMessageFrame): - await self._handle_message(frame) - else: - await self.push_frame(frame, direction) - - if isinstance(frame, TranscriptionFrame) or isinstance(frame, InterimTranscriptionFrame): - await self._handle_transcriptions(frame) - elif isinstance(frame, UserStartedSpeakingFrame) or isinstance(frame, UserStoppedSpeakingFrame): - await self._handle_interruptions(frame) + async def _push_transport_message(self, model: BaseModel, exclude_none: bool = True): + frame = TransportMessageFrame( + message=model.model_dump(exclude_none=exclude_none), + urgent=True) + await self.push_frame(frame) async def _handle_transcriptions(self, frame: Frame): - # TODO(aleix): Once we add support for using custom piplines, the STTs will - # be in the pipeline after this processor. This means the STT will have to - # push transcriptions upstream as well. + # TODO(aleix): Once we add support for using custom pipelines, the STTs will + # be in the pipeline after this processor. message = None if isinstance(frame, TranscriptionFrame): @@ -439,8 +431,7 @@ class RTVIProcessor(FrameProcessor): final=False)) if message: - frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) - await self.push_frame(frame) + await self._push_transport_message(message) async def _handle_interruptions(self, frame: Frame): message = None @@ -450,170 +441,150 @@ class RTVIProcessor(FrameProcessor): message = RTVIUserStoppedSpeakingMessage() if message: - frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) - await self.push_frame(frame) + await self._push_transport_message(message) + + async def _handle_bot_speaking(self, frame: Frame): + message = None + if isinstance(frame, BotStartedSpeakingFrame): + message = RTVIBotStartedSpeakingMessage() + elif isinstance(frame, BotStoppedSpeakingFrame): + message = RTVIBotStoppedSpeakingMessage() + + if message: + await self._push_transport_message(message) + + async def _message_task_handler(self): + while True: + try: + frame = await self._message_queue.get() + await self._handle_message(frame) + self._message_queue.task_done() + except asyncio.CancelledError: + break async def _handle_message(self, frame: TransportMessageFrame): try: message = RTVIMessage.model_validate(frame.message) except ValidationError as e: - await self._send_error(f"Invalid incoming message: {e}") + await self.send_error(f"Invalid incoming message: {e}") logger.warning(f"Invalid incoming message: {e}") return try: - success = True - error = None match message.type: - case "config-update": - await self._handle_config_update(RTVIConfig.model_validate(message.data)) - case "llm-get-context": - await self._handle_llm_get_context() - case "llm-append-context": - await self._handle_llm_append_context(RTVILLMContextData.model_validate(message.data)) - case "llm-update-context": - await self._handle_llm_update_context(RTVILLMContextData.model_validate(message.data)) - case "tts-speak": - await self._handle_tts_speak(RTVITTSSpeakData.model_validate(message.data)) - case "tts-interrupt": - await self._handle_tts_interrupt() - case _: - success = False - error = f"Unsupported type {message.type}" + case "describe-actions": + await self._handle_describe_actions(message.id) + case "describe-config": + await self._handle_describe_config(message.id) + case "get-config": + await self._handle_get_config(message.id) + case "update-config": + config = RTVIConfig.model_validate(message.data) + await self._handle_update_config(message.id, config) + case "action": + action = RTVIActionRun.model_validate(message.data) + await self._handle_action(message.id, action) + case "llm-function-call-result": + data = RTVILLMFunctionCallResultData.model_validate(message.data) + await self._handle_function_call_result(data) + + case _: + await self._send_error_response(message.id, f"Unsupported type {message.type}") - await self._send_response(message.id, success, error) except ValidationError as e: - await self._send_response(message.id, False, f"Invalid incoming message: {e}") + await self._send_error_response(message.id, f"Invalid incoming message: {e}") logger.warning(f"Invalid incoming message: {e}") except Exception as e: - await self._send_response(message.id, False, f"Exception processing message: {e}") + await self._send_error_response(message.id, f"Exception processing message: {e}") logger.warning(f"Exception processing message: {e}") - async def _handle_pipeline_setup(self, start_frame: StartFrame, config: RTVIConfig | None): - # TODO(aleix): We shouldn't need to save this in `self._tma_in`. - self._tma_in = LLMUserResponseAggregator() - tma_out = LLMAssistantResponseAggregator() + async def _handle_describe_config(self, request_id: str): + services = list(self._registered_services.values()) + message = RTVIDescribeConfig(id=request_id, data=RTVIDescribeConfigData(config=services)) + await self._push_transport_message(message) - llm_cls = self._registered_services["llm"].cls - llm_args = self._ctor_args["llm"] - llm = llm_cls(**llm_args) + async def _handle_describe_actions(self, request_id: str): + actions = list(self._registered_actions.values()) + message = RTVIDescribeActions(id=request_id, data=RTVIDescribeActionsData(actions=actions)) + await self._push_transport_message(message) - tts_cls = self._registered_services["tts"].cls - tts_args = self._ctor_args["tts"] - tts = tts_cls(**tts_args) + async def _handle_get_config(self, request_id: str): + message = RTVIConfigResponse(id=request_id, data=self._config) + await self._push_transport_message(message) - # TODO-CB: Eventually we'll need to switch the context aggregators to use the - # OpenAI context frames instead of message frames - context = OpenAILLMContext() - fc = FunctionCaller(context) + def _update_config_option(self, service: str, config: RTVIServiceOptionConfig): + for service_config in self._config.config: + if service_config.service == service: + for option_config in service_config.options: + if option_config.name == config.name: + option_config.value = config.value + return + # If we couldn't find a value for this config, we simply need to + # add it. + service_config.options.append(config) - tts_text = RTVITTSTextProcessor() - - pipeline = Pipeline([ - self._tma_in, - llm, - fc, - tts, - tts_text, - tma_out, - self._transport.output(), - ]) - - parent = self.get_parent() - if parent: - parent.link(pipeline) - - # We need to initialize the new pipeline with the same settings - # as the initial one. - start_frame = dataclasses.replace(start_frame) - await self.push_frame(start_frame) - - # Configure the pipeline - if config: - await self._handle_config_update(config) - - # Send new initial metrics with the new processors - processors = parent.processors_with_metrics() - processors.extend(pipeline.processors_with_metrics()) - ttfb = [{"processor": p.name, "value": 0.0} for p in processors] - processing = [{"processor": p.name, "value": 0.0} for p in processors] - tokens = [{"processor": p.name, "value": {"prompt_tokens": 0, - "completion_tokens": 0, - "total_tokens": 0}} for p in processors] - characters = [{"processor": p.name, "value": 0} for p in processors] - await self.push_frame(MetricsFrame(ttfb=ttfb, processing=processing, tokens=tokens, characters=characters)) - - self._pipeline = pipeline - - await self._maybe_send_bot_ready() - - async def _handle_config_service(self, config: RTVIServiceConfig): + async def _update_service_config(self, config: RTVIServiceConfig): service = self._registered_services[config.service] for option in config.options: handler = service._options_dict[option.name].handler - if handler: - await handler(self, option) + await handler(self, service.name, option) + self._update_config_option(service.name, option) - async def _handle_config_update(self, data: RTVIConfig): - for config in data.config: - await self._handle_config_service(config) + async def _update_config(self, data: RTVIConfig): + for service_config in data.config: + await self._update_service_config(service_config) - async def _handle_llm_get_context(self): - data = RTVILLMContextMessageData(messages=self._tma_in.messages) - message = RTVILLMContextMessage(data=data) - frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) + async def _handle_update_config(self, request_id: str, data: RTVIConfig): + # NOTE(aleix): The bot might be talking while we receive a new + # config. Let's interrupt it for now and update the config. Another + # solution is to wait until the bot stops speaking and then apply the + # config, but this definitely is more complicated to achieve. + await self.interrupt_bot() + await self._update_config(data) + await self._handle_get_config(request_id) + + async def _handle_function_call_result(self, data): + frame = FunctionCallResultFrame( + function_name=data.function_name, + tool_call_id=data.tool_call_id, + arguments=data.arguments, + result=data.result) await self.push_frame(frame) - async def _handle_llm_append_context(self, data: RTVILLMContextData): - if data and data.messages: - frame = LLMMessagesAppendFrame(data.messages) - await self.push_frame(frame) + async def _handle_action(self, request_id: str, data: RTVIActionRun): + action_id = self._action_id(data.service, data.action) + if action_id not in self._registered_actions: + await self._send_error_response(request_id, f"Action {action_id} not registered") + return + action = self._registered_actions[action_id] + arguments = {} + if data.arguments: + for arg in data.arguments: + arguments[arg.name] = arg.value + result = await action.handler(self, action.service, arguments) + message = RTVIActionResponse(id=request_id, data=RTVIActionResponseData(result=result)) + await self._push_transport_message(message) - async def _handle_llm_update_context(self, data: RTVILLMContextData): - if data and data.messages: - frame = LLMMessagesUpdateFrame(data.messages) - await self.push_frame(frame) - - async def _handle_tts_speak(self, data: RTVITTSSpeakData): - if data and data.text: - if data.interrupt: - await self._handle_tts_interrupt() - frame = TTSSpeakFrame(text=data.text) - await self.push_frame(frame) - - async def _handle_tts_interrupt(self): - await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) - - async def _on_first_participant_joined(self, transport, participant): - self._first_participant_joined = True - await self._maybe_send_bot_ready() + async def _transport_on_joined(self, transport, participant): + self._transport_joined = True async def _maybe_send_bot_ready(self): - if self._pipeline and self._first_participant_joined: - message = RTVIBotReady() - frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) - await self.push_frame(frame) + if self._pipeline_started and self._transport_joined: + await self._send_bot_ready() - async def _send_error(self, error: str): - message = RTVIError(data=RTVIErrorData(message=error)) - frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) - await self.push_frame(frame) + async def _send_bot_ready(self): + if not self._params.send_bot_ready: + return - async def _send_response(self, id: str, success: bool, error: str | None = None): - # TODO(aleix): This is a bit hacky, but we might get invalid - # configuration or something might going wrong during setup and we would - # like to send the error to the client. However, if the pipeline is not - # setup yet we don't have an output transport and therefore we can't - # send any messages. So, we setup a super basic pipeline with just the - # output transport so we can send messages. - if not self._pipeline: - pipeline = Pipeline([self._transport.output()]) - self._pipeline = pipeline + message = RTVIBotReady( + data=RTVIBotReadyData( + version=RTVI_PROTOCOL_VERSION, + config=self._config.config)) + await self._push_transport_message(message) - parent = self.get_parent() - if parent: - parent.link(pipeline) + async def _send_error_response(self, id: str, error: str): + message = RTVIErrorResponse(id=id, data=RTVIErrorResponseData(error=error)) + await self._push_transport_message(message) - message = RTVIResponse(id=id, data=RTVIResponseData(success=success, error=error)) - frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) - await self.push_frame(frame) + def _action_id(self, service: str, action: str) -> str: + return f"{service}:{action}" diff --git a/src/pipecat/processors/logger.py b/src/pipecat/processors/logger.py index 6f07548af..79334ba73 100644 --- a/src/pipecat/processors/logger.py +++ b/src/pipecat/processors/logger.py @@ -4,7 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # -from pipecat.frames.frames import Frame +from pipecat.frames.frames import BotSpeakingFrame, Frame, AudioRawFrame, TransportMessageFrame from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from loguru import logger from typing import Optional @@ -12,16 +12,25 @@ logger = logger.opt(ansi=True) class FrameLogger(FrameProcessor): - def __init__(self, prefix="Frame", color: Optional[str] = None): + def __init__( + self, + prefix="Frame", + color: Optional[str] = None, + ignored_frame_types: Optional[list] = [ + BotSpeakingFrame, + AudioRawFrame, + TransportMessageFrame]): super().__init__() self._prefix = prefix self._color = color + self._ignored_frame_types = tuple(ignored_frame_types) if ignored_frame_types else None async def process_frame(self, frame: Frame, direction: FrameDirection): - dir = "<" if direction is FrameDirection.UPSTREAM else ">" - msg = f"{dir} {self._prefix}: {frame}" - if self._color: - msg = f"<{self._color}>{msg}" - logger.debug(msg) + if self._ignored_frame_types and not isinstance(frame, self._ignored_frame_types): + dir = "<" if direction is FrameDirection.UPSTREAM else ">" + msg = f"{dir} {self._prefix}: {frame}" + if self._color: + msg = f"<{self._color}>{msg}" + logger.debug(msg) await self.push_frame(frame, direction) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 2b828df31..33abf4e15 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -20,34 +20,16 @@ from pipecat.frames.frames import ( StartFrame, StartInterruptionFrame, TTSSpeakFrame, - TTSStartedFrame, - TTSStoppedFrame, TTSVoiceUpdateFrame, TextFrame, - VisionImageRawFrame, + VisionImageRawFrame ) from pipecat.processors.async_frame_processor import AsyncFrameProcessor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.utils.audio import calculate_audio_volume +from pipecat.utils.string import match_endofsentence from pipecat.utils.utils import exp_smoothing -import re - - -ENDOFSENTENCE_PATTERN_STR = r""" - (? bool: - return ENDOFSENTENCE_PATTERN.search(text.rstrip()) is not None +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext class AIService(FrameProcessor): @@ -115,27 +97,51 @@ class LLMService(AIService): self._start_callbacks = {} # TODO-CB: callback function type - def register_function(self, function_name: str, callback, start_callback=None): + def register_function(self, function_name: str | None, callback, start_callback=None): + # 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? if start_callback: self._start_callbacks[function_name] = start_callback - def unregister_function(self, function_name: str): + def unregister_function(self, function_name: str | None): del self._callbacks[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(): + return True return function_name in self._callbacks.keys() - async def call_function(self, function_name: str, args): + async def call_function( + self, + *, + context: OpenAILLMContext, + tool_call_id: str, + function_name: str, + arguments: str) -> None: + f = None if function_name in self._callbacks.keys(): - return await self._callbacks[function_name](self, args) - return None + f = self._callbacks[function_name] + elif None in self._callbacks.keys(): + f = self._callbacks[None] + else: + return None + await context.call_function( + f, + function_name=function_name, + tool_call_id=tool_call_id, + arguments=arguments, + llm=self) + # QUESTION FOR CB: maybe this isn't needed anymore? async def call_start_function(self, function_name: str): if function_name in self._start_callbacks.keys(): await self._start_callbacks[function_name](self) + elif None in self._start_callbacks.keys(): + return await self._start_callbacks[None](function_name) class TTSService(AIService): @@ -185,11 +191,9 @@ class TTSService(AIService): if not text: return - await self.push_frame(TTSStartedFrame()) await self.start_processing_metrics() await self.process_generator(self.run_tts(text)) await self.stop_processing_metrics() - await self.push_frame(TTSStoppedFrame()) if self._push_text_frames: # We send the original text after the audio. This way, if we are # interrupted, the text is not added to the assistant context. diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 7854bb792..7974bd3e0 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -5,19 +5,40 @@ # import base64 +import json +import io +import copy +from typing import List, Optional +from dataclasses import dataclass +from PIL import Image +from asyncio import CancelledError +import re from pipecat.frames.frames import ( Frame, + LLMEnablePromptCachingFrame, LLMModelUpdateFrame, TextFrame, VisionImageRawFrame, + UserImageRequestFrame, + UserImageRawFrame, LLMMessagesFrame, LLMFullResponseStartFrame, - LLMFullResponseEndFrame + LLMFullResponseEndFrame, + FunctionCallResultFrame, + FunctionCallInProgressFrame, + StartInterruptionFrame ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_services import LLMService -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext, OpenAILLMContextFrame +from pipecat.processors.aggregators.openai_llm_context import ( + OpenAILLMContext, + OpenAILLMContextFrame +) +from pipecat.processors.aggregators.llm_response import ( + LLMUserContextAggregator, + LLMAssistantContextAggregator +) from loguru import logger @@ -26,87 +47,95 @@ try: except ModuleNotFoundError as e: logger.error(f"Exception: {e}") logger.error( - "In order to use Anthropic, you need to `pip install pipecat-ai[anthropic]`. Also, set `ANTHROPIC_API_KEY` environment variable.") + "In order to use Anthropic, you need to `pip install pipecat-ai[anthropic]`. " + + "Also, set `ANTHROPIC_API_KEY` environment variable.") raise Exception(f"Missing module: {e}") +@dataclass +class AnthropicImageMessageFrame(Frame): + user_image_raw_frame: UserImageRawFrame + text: Optional[str] = None + + +@dataclass +class AnthropicContextAggregatorPair: + _user: 'AnthropicUserContextAggregator' + _assistant: 'AnthropicAssistantContextAggregator' + + def user(self) -> 'AnthropicUserContextAggregator': + return self._user + + def assistant(self) -> 'AnthropicAssistantContextAggregator': + return self._assistant + + class AnthropicLLMService(LLMService): """This class implements inference with Anthropic's AI models - - This service translates internally from OpenAILLMContext to the messages format - expected by the Anthropic Python SDK. We are using the OpenAILLMContext as a lingua - franca for all LLM services, so that it is easy to switch between different LLMs. """ def __init__( self, *, api_key: str, - model: str = "claude-3-opus-20240229", - max_tokens: int = 1024): - super().__init__() + model: str = "claude-3-5-sonnet-20240620", + max_tokens: int = 4096, + enable_prompt_caching_beta: bool = False, + **kwargs): + super().__init__(**kwargs) self._client = AsyncAnthropic(api_key=api_key) self._model = model self._max_tokens = max_tokens + self._enable_prompt_caching_beta = enable_prompt_caching_beta def can_generate_metrics(self) -> bool: return True - def _get_messages_from_openai_context( - self, context: OpenAILLMContext): - openai_messages = context.get_messages() - anthropic_messages = [] + @property + def enable_prompt_caching_beta(self) -> bool: + return self._enable_prompt_caching_beta - for message in openai_messages: - role = message["role"] - text = message["content"] - if role == "system": - role = "user" - if message.get("mime_type") == "image/jpeg": - # vision frame - encoded_image = base64.b64encode(message["data"].getvalue()).decode("utf-8") - anthropic_messages.append({ - "role": role, - "content": [{ - "type": "image", - "source": { - "type": "base64", - "media_type": message.get("mime_type"), - "data": encoded_image, - } - }, { - "type": "text", - "text": text - }] - }) - else: - # Text frame. Anthropic needs the roles to alternate. This will - # cause an issue with interruptions. So, if we detect we are the - # ones asking again it probably means we were interrupted. - if role == "user" and len(anthropic_messages) > 1: - last_message = anthropic_messages[-1] - if last_message["role"] == "user": - anthropic_messages = anthropic_messages[:-1] - content = last_message["content"] - anthropic_messages.append( - {"role": "user", "content": f"Sorry, I just asked you about [{content}] but now I would like to know [{text}]."}) - else: - anthropic_messages.append({"role": role, "content": text}) - else: - anthropic_messages.append({"role": role, "content": text}) - - return anthropic_messages + @staticmethod + def create_context_aggregator(context: OpenAILLMContext) -> AnthropicContextAggregatorPair: + user = AnthropicUserContextAggregator(context) + assistant = AnthropicAssistantContextAggregator(user) + return AnthropicContextAggregatorPair( + _user=user, + _assistant=assistant + ) async def _process_context(self, context: OpenAILLMContext): - await self.push_frame(LLMFullResponseStartFrame()) - try: - logger.debug(f"Generating chat: {context.get_messages_json()}") + # Usage tracking. We track the usage reported by Anthropic in prompt_tokens and + # completion_tokens. We also estimate the completion tokens from output text + # and use that estimate if we are interrupted, because we almost certainly won't + # get a complete usage report if the task we're running in is cancelled. + prompt_tokens = 0 + completion_tokens = 0 + completion_tokens_estimate = 0 + use_completion_tokens_estimate = False + cache_creation_input_tokens = 0 + cache_read_input_tokens = 0 - messages = self._get_messages_from_openai_context(context) + try: + await self.push_frame(LLMFullResponseStartFrame()) + await self.start_processing_metrics() + + logger.debug( + f"Generating chat: {context.system} | {context.get_messages_for_logging()}") + + messages = context.messages + if self._enable_prompt_caching_beta: + messages = context.get_messages_with_cache_control_markers() + + api_call = self._client.messages.create + if self._enable_prompt_caching_beta: + api_call = self._client.beta.prompt_caching.messages.create await self.start_ttfb_metrics() - response = await self._client.messages.create( + response = await api_call( + tools=context.tools or [], + system=context.system or [], messages=messages, model=self._model, max_tokens=self._max_tokens, @@ -114,32 +143,397 @@ class AnthropicLLMService(LLMService): await self.stop_ttfb_metrics() + # Function calling + tool_use_block = None + json_accumulator = '' + async for event in response: # logger.debug(f"Anthropic LLM event: {event}") - if (event.type == "content_block_delta"): - await self.push_frame(TextFrame(event.delta.text)) + # Aggregate streaming content, create frames, trigger events + + if (event.type == "content_block_delta"): + if hasattr(event.delta, 'text'): + await self.push_frame(TextFrame(event.delta.text)) + completion_tokens_estimate += self._estimate_tokens(event.delta.text) + elif hasattr(event.delta, 'partial_json') and tool_use_block: + json_accumulator += event.delta.partial_json + completion_tokens_estimate += self._estimate_tokens( + event.delta.partial_json) + elif (event.type == "content_block_start"): + if event.content_block.type == "tool_use": + tool_use_block = event.content_block + json_accumulator = '' + elif ((event.type == "message_delta" and + hasattr(event.delta, 'stop_reason') + and event.delta.stop_reason == 'tool_use')): + if tool_use_block: + await self.call_function(context=context, + tool_call_id=tool_use_block.id, + function_name=tool_use_block.name, + arguments=json.loads(json_accumulator)) + + # Calculate usage. Do this here in its own if statement, because there may be usage + # data embedded in messages that we do other processing for, above. + if hasattr(event, "usage"): + prompt_tokens += event.usage.input_tokens if hasattr( + event.usage, "input_tokens") else 0 + completion_tokens += event.usage.output_tokens if hasattr( + event.usage, "output_tokens") else 0 + elif hasattr(event, "message") and hasattr(event.message, "usage"): + prompt_tokens += event.message.usage.input_tokens if hasattr( + event.message.usage, "input_tokens") else 0 + completion_tokens += event.message.usage.output_tokens if hasattr( + event.message.usage, "output_tokens") else 0 + if hasattr(event.message.usage, "cache_creation_input_tokens"): + cache_creation_input_tokens += event.message.usage.cache_creation_input_tokens + logger.debug(f"Cache creation input tokens: {cache_creation_input_tokens}") + if hasattr(event.message.usage, "cache_read_input_tokens"): + cache_read_input_tokens += event.message.usage.cache_read_input_tokens + logger.debug(f"Cache read input tokens: {cache_read_input_tokens}") + total_input_tokens = prompt_tokens + cache_creation_input_tokens + cache_read_input_tokens + if total_input_tokens >= 1024: + context.turns_above_cache_threshold += 1 + + except CancelledError: + # If we're interrupted, we won't get a complete usage report. So set our flag to use the + # token estimate. The reraise the exception so all the processors running in this task + # also get cancelled. + use_completion_tokens_estimate = True + raise except Exception as e: logger.exception(f"{self} exception: {e}") finally: + await self.stop_processing_metrics() await self.push_frame(LLMFullResponseEndFrame()) + comp_tokens = completion_tokens if not use_completion_tokens_estimate else completion_tokens_estimate + await self._report_usage_metrics( + prompt_tokens=prompt_tokens, + completion_tokens=comp_tokens, + cache_creation_input_tokens=cache_creation_input_tokens, + cache_read_input_tokens=cache_read_input_tokens + ) async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) context = None - if isinstance(frame, OpenAILLMContextFrame): - context: OpenAILLMContext = frame.context + context = frame.context elif isinstance(frame, LLMMessagesFrame): - context = OpenAILLMContext.from_messages(frame.messages) + context = AnthropicLLMContext.from_messages(frame.messages) elif isinstance(frame, VisionImageRawFrame): - context = OpenAILLMContext.from_image_frame(frame) + # This is only useful in very simple pipelines because it creates + # a new context. Generally we want a context manager to catch + # UserImageRawFrames coming through the pipeline and add them + # to the context. + context = AnthropicLLMContext.from_image_frame(frame) elif isinstance(frame, LLMModelUpdateFrame): logger.debug(f"Switching LLM model to: [{frame.model}]") self._model = frame.model + elif isinstance(frame, LLMEnablePromptCachingFrame): + logger.debug(f"Setting enable prompt caching to: [{frame.enable}]") + self._enable_prompt_caching_beta = frame.enable else: await self.push_frame(frame, direction) if context: await self._process_context(context) + + async def request_image_frame(self, user_id: str, *, text_content: str = None): + await self.push_frame(UserImageRequestFrame(user_id=user_id, context=text_content), + FrameDirection.UPSTREAM) + + def _estimate_tokens(self, text: str) -> int: + return int(len(re.split(r'[^\w]+', text)) * 1.3) + + async def _report_usage_metrics( + self, + prompt_tokens: int, + completion_tokens: int, + cache_creation_input_tokens: int, + cache_read_input_tokens: int): + if prompt_tokens or completion_tokens or cache_creation_input_tokens or cache_read_input_tokens: + tokens = { + "processor": self.name, + "model": self._model, + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "cache_creation_input_tokens": cache_creation_input_tokens, + "cache_read_input_tokens": cache_read_input_tokens, + "total_tokens": prompt_tokens + completion_tokens + } + await self.start_llm_usage_metrics(tokens) + + +class AnthropicLLMContext(OpenAILLMContext): + def __init__( + self, + messages: list[dict] | None = None, + tools: list[dict] | None = None, + tool_choice: dict | None = None, + *, + system: List | None = None + ): + super().__init__(messages=messages, tools=tools, tool_choice=tool_choice) + self._user_image_request_context = {} + + # For beta prompt caching. This is a counter that tracks the number of turns + # we've seen above the cache threshold. We reset this when we reset the + # messages list. We only care about this number being 0, 1, or 2. But + # it's easiest just to treat it as a counter. + self.turns_above_cache_threshold = 0 + + self.system = system + + @classmethod + def from_openai_context(cls, openai_context: OpenAILLMContext): + self = cls( + messages=openai_context.messages, + tools=openai_context.tools, + tool_choice=openai_context.tool_choice, + ) + self._restructure_from_openai_messages() + return self + + @classmethod + def from_messages(cls, messages: List[dict]) -> "AnthropicLLMContext": + self = cls(messages=messages) + self._restructure_from_openai_messages() + return self + + @classmethod + def from_image_frame(cls, frame: VisionImageRawFrame) -> "AnthropicLLMContext": + context = cls() + context.add_image_frame_message( + format=frame.format, + size=frame.size, + image=frame.image, + text=frame.text) + return context + + def set_messages(self, messages: List): + self.turns_above_cache_threshold = 0 + self._messages[:] = messages + self._restructure_from_openai_messages() + + def add_image_frame_message( + self, *, format: str, size: tuple[int, int], image: bytes, text: str = None): + buffer = io.BytesIO() + Image.frombytes(format, size, image).save(buffer, format="JPEG") + encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8") + # Anthropic docs say that the image should be the first content block in the message. + content = [{"type": "image", + "source": { + "type": "base64", + "media_type": "image/jpeg", + "data": encoded_image, + }}] + if text: + content.append({"type": "text", "text": text}) + self.add_message({"role": "user", "content": content}) + + def add_message(self, message): + try: + if self.messages: + # Anthropic requires that roles alternate. If this message's role is the same as the + # last message, we should add this message's content to the last message. + if self.messages[-1]["role"] == message["role"]: + # if the last message has just a content string, convert it to a list + # in the proper format + if isinstance(self.messages[-1]["content"], str): + self.messages[-1]["content"] = [{"type": "text", + "text": self.messages[-1]["content"]}] + # if this message has just a content string, convert it to a list + # in the proper format + if isinstance(message["content"], str): + message["content"] = [{"type": "text", "text": message["content"]}] + # append the content of this message to the last message + self.messages[-1]["content"].extend(message["content"]) + else: + self.messages.append(message) + else: + self.messages.append(message) + except Exception as e: + logger.error(f"Error adding message: {e}") + + def get_messages_with_cache_control_markers(self) -> List[dict]: + try: + messages = copy.deepcopy(self.messages) + if self.turns_above_cache_threshold >= 1 and messages[-1]["role"] == "user": + if isinstance(messages[-1]["content"], str): + messages[-1]["content"] = [{"type": "text", "text": messages[-1]["content"]}] + messages[-1]["content"][-1]["cache_control"] = {"type": "ephemeral"} + if (self.turns_above_cache_threshold >= 2 and + len(messages) > 2 and messages[-3]["role"] == "user"): + if isinstance(messages[-3]["content"], str): + messages[-3]["content"] = [{"type": "text", "text": messages[-3]["content"]}] + messages[-3]["content"][-1]["cache_control"] = {"type": "ephemeral"} + return messages + except Exception as e: + logger.error(f"Error adding cache control marker: {e}") + return self.messages + + def _restructure_from_openai_messages(self): + # See if we should pull the system message out of our context.messages list. (For + # compatibility with Open AI messages format.) + if self.messages and self.messages[0]["role"] == "system": + if len(self.messages) == 1: + # If we have only have a system message in the list, all we can really do + # without introducing too much magic is change the role to "user". + self.messages[0]["role"] = "user" + else: + # If we have more than one message, we'll pull the system message out of the + # list. + self.system = self.messages[0]["content"] + self.messages.pop(0) + + def get_messages_for_logging(self) -> str: + msgs = [] + for message in self.messages: + msg = copy.deepcopy(message) + if "content" in msg: + if isinstance(msg["content"], list): + for item in msg["content"]: + if item["type"] == "image": + item["source"]["data"] = "..." + msgs.append(msg) + return json.dumps(msgs) + + +class AnthropicUserContextAggregator(LLMUserContextAggregator): + def __init__(self, context: OpenAILLMContext | AnthropicLLMContext): + super().__init__(context=context) + + if isinstance(context, OpenAILLMContext): + self._context = AnthropicLLMContext.from_openai_context(context) + + 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}") + +# +# Claude returns a text content block along with a tool use content block. This works quite nicely +# with streaming. We get the text first, so we can start streaming it right away. Then we get the +# tool_use block. While the text is streaming to TTS and the transport, we can run the tool call. +# +# But Claude is verbose. It would be nice to come up with prompt language that suppresses Claude's +# chattiness about it's tool thinking. +# + + +class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator): + def __init__(self, user_context_aggregator: AnthropicUserContextAggregator): + super().__init__(context=user_context_aggregator._context) + self._user_context_aggregator = user_context_aggregator + self._function_call_in_progress = None + self._function_call_result = None + + 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 + 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): + try: + 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) + await self._user_context_aggregator.push_context_frame() + except Exception as e: + logger.error(f"Error processing AnthropicImageMessageFrame: {e}") + + def add_message(self, message): + self._user_context_aggregator.add_message(message) + + async def _push_aggregation(self): + if not self._aggregation: + return + + run_llm = False + + aggregation = self._aggregation + self._aggregation = "" + + try: + if self._function_call_result: + frame = self._function_call_result + self._function_call_result = None + self._context.add_message({ + "role": "assistant", + "content": [ + { + "type": "text", + "text": aggregation + }, + { + "type": "tool_use", + "id": frame.tool_call_id, + "name": frame.function_name, + "input": frame.arguments + } + ] + }) + self._context.add_message({ + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": frame.tool_call_id, + "content": json.dumps(frame.result) + } + ] + }) + run_llm = True + else: + self._context.add_message({"role": "assistant", "content": aggregation}) + + if run_llm: + await self._user_context_aggregator.push_context_frame() + + except Exception as e: + logger.error(f"Error processing frame: {e}") diff --git a/src/pipecat/services/azure.py b/src/pipecat/services/azure.py index 6e81c881d..76e884992 100644 --- a/src/pipecat/services/azure.py +++ b/src/pipecat/services/azure.py @@ -17,9 +17,10 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, - MetricsFrame, StartFrame, SystemFrame, + TTSStartedFrame, + TTSStoppedFrame, TranscriptionFrame, URLImageRawFrame) from pipecat.processors.frame_processor import FrameDirection @@ -106,8 +107,10 @@ class AzureTTSService(TTSService): if result.reason == ResultReason.SynthesizingAudioCompleted: await self.start_tts_usage_metrics(text) await self.stop_ttfb_metrics() + await self.push_frame(TTSStartedFrame()) # Azure always sends a 44-byte header. Strip it off. yield AudioRawFrame(audio=result.audio_data[44:], sample_rate=16000, num_channels=1) + await self.push_frame(TTSStoppedFrame()) elif result.reason == ResultReason.Canceled: cancellation_details = result.cancellation_details logger.warning(f"Speech synthesis canceled: {cancellation_details.reason}") diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 5233366cd..735d12b5b 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -15,13 +15,15 @@ from typing import AsyncGenerator from pipecat.processors.frame_processor import FrameDirection from pipecat.frames.frames import ( CancelFrame, + ErrorFrame, Frame, AudioRawFrame, StartInterruptionFrame, StartFrame, EndFrame, + TTSStartedFrame, + TTSStoppedFrame, TextFrame, - MetricsFrame, LLMFullResponseEndFrame ) from pipecat.services.ai_services import TTSService @@ -153,6 +155,7 @@ class CartesiaTTSService(TTSService): continue if msg["type"] == "done": await self.stop_ttfb_metrics() + await self.push_frame(TTSStoppedFrame()) # Unset _context_id but not the _context_id_start_timestamp # because we are likely still playing out audio and need the # timestamp to set send context frames. @@ -173,6 +176,13 @@ class CartesiaTTSService(TTSService): num_channels=1 ) await self.push_frame(frame) + elif msg["type"] == "error": + logger.error(f"{self} error: {msg}") + await self.push_frame(TTSStoppedFrame()) + await self.stop_all_metrics() + await self.push_error(ErrorFrame(f'{self} error: {msg["error"]}')) + else: + logger.error(f"Cartesia error, unknown message type: {msg}") except asyncio.CancelledError: pass except Exception as e: @@ -207,6 +217,7 @@ class CartesiaTTSService(TTSService): await self._connect() if not self._context_id: + await self.push_frame(TTSStartedFrame()) await self.start_ttfb_metrics() self._context_id = str(uuid.uuid4()) @@ -227,7 +238,8 @@ class CartesiaTTSService(TTSService): await self._websocket.send(json.dumps(msg)) await self.start_tts_usage_metrics(text) except Exception as e: - logger.exception(f"{self} error sending message: {e}") + logger.error(f"{self} error sending message: {e}") + await self.push_frame(TTSStoppedFrame()) await self._disconnect() await self._connect() return diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index 8d58def56..035fdd25c 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -15,9 +15,10 @@ from pipecat.frames.frames import ( ErrorFrame, Frame, InterimTranscriptionFrame, - MetricsFrame, StartFrame, SystemFrame, + TTSStartedFrame, + TTSStoppedFrame, TranscriptionFrame) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_services import AsyncAIService, TTSService @@ -96,10 +97,12 @@ class DeepgramTTSService(TTSService): await self.start_tts_usage_metrics(text) + await self.push_frame(TTSStartedFrame()) async for data in r.content: await self.stop_ttfb_metrics() frame = AudioRawFrame(audio=data, sample_rate=self._sample_rate, num_channels=1) yield frame + await self.push_frame(TTSStoppedFrame()) except Exception as e: logger.exception(f"{self} exception: {e}") diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 0de629034..974619ea8 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -9,7 +9,7 @@ import aiohttp from typing import AsyncGenerator, Literal from pydantic import BaseModel -from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame, MetricsFrame +from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame, TTSStartedFrame, TTSStoppedFrame from pipecat.services.ai_services import TTSService from loguru import logger @@ -70,8 +70,10 @@ class ElevenLabsTTSService(TTSService): await self.start_tts_usage_metrics(text) + await self.push_frame(TTSStartedFrame()) async for chunk in r.content: if len(chunk) > 0: await self.stop_ttfb_metrics() frame = AudioRawFrame(chunk, 16000, 1) yield frame + await self.push_frame(TTSStoppedFrame()) diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index ad15b9907..52da196d2 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -9,6 +9,7 @@ import base64 import io import json import httpx +from dataclasses import dataclass from typing import AsyncGenerator, List, Literal @@ -23,11 +24,17 @@ from pipecat.frames.frames import ( LLMFullResponseStartFrame, LLMMessagesFrame, LLMModelUpdateFrame, - MetricsFrame, + TTSStartedFrame, + TTSStoppedFrame, TextFrame, URLImageRawFrame, - VisionImageRawFrame + VisionImageRawFrame, + FunctionCallResultFrame, + FunctionCallInProgressFrame, + StartInterruptionFrame ) +from pipecat.processors.aggregators.llm_response import LLMUserContextAggregator, LLMAssistantContextAggregator + from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContext, OpenAILLMContextFrame @@ -41,12 +48,7 @@ from pipecat.services.ai_services import ( try: from openai import AsyncOpenAI, AsyncStream, DefaultAsyncHttpxClient, BadRequestError - from openai.types.chat import ( - ChatCompletionChunk, - ChatCompletionFunctionMessageParam, - ChatCompletionMessageParam, - ChatCompletionToolParam - ) + from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam except ModuleNotFoundError as e: logger.error(f"Exception: {e}") logger.error( @@ -137,6 +139,7 @@ class BaseOpenAILLMService(LLMService): if chunk.usage: tokens = { "processor": self.name, + "model": self._model, "prompt_tokens": chunk.usage.prompt_tokens, "completion_tokens": chunk.usage.completion_tokens, "total_tokens": chunk.usage.total_tokens @@ -190,44 +193,12 @@ class BaseOpenAILLMService(LLMService): arguments ): arguments = json.loads(arguments) - result = await self.call_function(function_name, arguments) - arguments = json.dumps(arguments) - if isinstance(result, (str, dict)): - # Handle it in "full magic mode" - tool_call = ChatCompletionFunctionMessageParam({ - "role": "assistant", - "tool_calls": [ - { - "id": tool_call_id, - "function": { - "arguments": arguments, - "name": function_name - }, - "type": "function" - } - ] - - }) - context.add_message(tool_call) - if isinstance(result, dict): - result = json.dumps(result) - tool_result = ChatCompletionToolParam({ - "tool_call_id": tool_call_id, - "role": "tool", - "content": result - }) - context.add_message(tool_result) - # re-prompt to get a human answer - await self._process_context(context) - elif isinstance(result, list): - # reduced magic - for msg in result: - context.add_message(msg) - await self._process_context(context) - elif isinstance(result, type(None)): - pass - else: - raise TypeError(f"Unknown return type from function callback: {type(result)}") + await self.call_function( + context=context, + tool_call_id=tool_call_id, + function_name=function_name, + arguments=arguments + ) async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -253,11 +224,32 @@ class BaseOpenAILLMService(LLMService): await self.push_frame(LLMFullResponseEndFrame()) +@dataclass +class OpenAIContextAggregatorPair: + _user: 'OpenAIUserContextAggregator' + _assistant: 'OpenAIAssistantContextAggregator' + + def user(self) -> 'OpenAIUserContextAggregator': + return self._user + + def assistant(self) -> 'OpenAIAssistantContextAggregator': + return self._assistant + + class OpenAILLMService(BaseOpenAILLMService): def __init__(self, *, model: str = "gpt-4o", **kwargs): super().__init__(model=model, **kwargs) + @staticmethod + def create_context_aggregator(context: OpenAILLMContext) -> OpenAIContextAggregatorPair: + user = OpenAIUserContextAggregator(context) + assistant = OpenAIAssistantContextAggregator(user) + return OpenAIContextAggregatorPair( + _user=user, + _assistant=assistant + ) + class OpenAIImageGenService(ImageGenService): @@ -352,10 +344,89 @@ class OpenAITTSService(TTSService): await self.start_tts_usage_metrics(text) + await self.push_frame(TTSStartedFrame()) async for chunk in r.iter_bytes(8192): if len(chunk) > 0: await self.stop_ttfb_metrics() frame = AudioRawFrame(chunk, 24_000, 1) yield frame + await self.push_frame(TTSStoppedFrame()) except BadRequestError as e: logger.exception(f"{self} error generating TTS: {e}") + + +class OpenAIUserContextAggregator(LLMUserContextAggregator): + def __init__(self, context: OpenAILLMContext): + super().__init__(context=context) + + +class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): + def __init__(self, user_context_aggregator: OpenAIUserContextAggregator): + super().__init__(context=user_context_aggregator._context) + self._user_context_aggregator = user_context_aggregator + self._function_call_in_progress = None + self._function_call_result = None + + 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 + # TODO-CB: Kwin wants us to refactor this out of here but I REFUSE + await self._push_aggregation() + else: + logger.warning( + f"FunctionCallResultFrame tool_call_id does not match FunctionCallInProgressFrame tool_call_id") + self._function_call_in_progress = None + self._function_call_result = None + + def add_message(self, message): + self._user_context_aggregator.add_message(message) + + async def _push_aggregation(self): + if not (self._aggregation or self._function_call_result): + return + + run_llm = False + + aggregation = self._aggregation + self._aggregation = "" + + try: + if self._function_call_result: + frame = self._function_call_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 + }) + self._function_call_result = None + run_llm = True + else: + self._context.add_message({"role": "assistant", "content": aggregation}) + + if run_llm: + await self._user_context_aggregator.push_context_frame() + + except Exception as e: + logger.error(f"Error processing frame: {e}") diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index b738040b6..2f4ae9851 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -9,7 +9,7 @@ import struct from typing import AsyncGenerator -from pipecat.frames.frames import AudioRawFrame, Frame, MetricsFrame +from pipecat.frames.frames import AudioRawFrame, Frame, TTSStartedFrame, TTSStoppedFrame from pipecat.services.ai_services import TTSService from loguru import logger @@ -62,6 +62,7 @@ class PlayHTTTSService(TTSService): await self.start_tts_usage_metrics(text) + await self.push_frame(TTSStartedFrame()) async for chunk in playht_gen: # skip the RIFF header. if in_header: @@ -81,5 +82,6 @@ class PlayHTTTSService(TTSService): await self.stop_ttfb_metrics() frame = AudioRawFrame(chunk, 16000, 1) yield frame + await self.push_frame(TTSStoppedFrame()) except Exception as e: logger.exception(f"{self} error generating TTS: {e}") diff --git a/src/pipecat/services/together.py b/src/pipecat/services/together.py new file mode 100644 index 000000000..685609858 --- /dev/null +++ b/src/pipecat/services/together.py @@ -0,0 +1,314 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import base64 +import json +import io +import copy +from typing import List, Optional +from dataclasses import dataclass +from asyncio import CancelledError +import re +import uuid + +from pipecat.frames.frames import ( + Frame, + LLMModelUpdateFrame, + TextFrame, + VisionImageRawFrame, + UserImageRequestFrame, + UserImageRawFrame, + LLMMessagesFrame, + LLMFullResponseStartFrame, + LLMFullResponseEndFrame, + FunctionCallResultFrame, + FunctionCallInProgressFrame, + StartInterruptionFrame +) +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.ai_services import LLMService +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext, OpenAILLMContextFrame +from pipecat.processors.aggregators.llm_response import LLMUserContextAggregator, LLMAssistantContextAggregator + +from loguru import logger + +try: + from together import AsyncTogether +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error( + "In order to use Together.ai, you need to `pip install pipecat-ai[together]`. Also, set `TOGETHER_API_KEY` environment variable.") + raise Exception(f"Missing module: {e}") + + +@dataclass +class TogetherContextAggregatorPair: + _user: 'TogetherUserContextAggregator' + _assistant: 'TogetherAssistantContextAggregator' + + def user(self) -> 'TogetherUserContextAggregator': + return self._user + + def assistant(self) -> 'TogetherAssistantContextAggregator': + return self._assistant + + +class TogetherLLMService(LLMService): + """This class implements inference with Together's Llama 3.1 models + """ + + def __init__( + self, + *, + api_key: str, + model: str = "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo", + max_tokens: int = 4096, + **kwargs): + super().__init__(**kwargs) + self._client = AsyncTogether(api_key=api_key) + self._model = model + self._max_tokens = max_tokens + + def can_generate_metrics(self) -> bool: + return True + + @staticmethod + def create_context_aggregator(context: OpenAILLMContext) -> TogetherContextAggregatorPair: + user = TogetherUserContextAggregator(context) + assistant = TogetherAssistantContextAggregator(user) + return TogetherContextAggregatorPair( + _user=user, + _assistant=assistant + ) + + async def _process_context(self, context: OpenAILLMContext): + try: + await self.push_frame(LLMFullResponseStartFrame()) + await self.start_processing_metrics() + + logger.debug(f"Generating chat: {context.get_messages_for_logging()}") + + await self.start_ttfb_metrics() + + stream = await self._client.chat.completions.create( + messages=context.messages, + model=self._model, + max_tokens=self._max_tokens, + stream=True, + ) + + # Function calling + got_first_chunk = False + accumulating_function_call = False + function_call_accumulator = "" + + async for chunk in stream: + # logger.debug(f"Together LLM event: {chunk}") + if chunk.usage: + tokens = { + "processor": self.name, + "model": self._model, + "prompt_tokens": chunk.usage.prompt_tokens, + "completion_tokens": chunk.usage.completion_tokens, + "total_tokens": chunk.usage.total_tokens + } + await self.start_llm_usage_metrics(tokens) + + if len(chunk.choices) == 0: + continue + + if not got_first_chunk: + await self.stop_ttfb_metrics() + if chunk.choices[0].delta.content: + got_first_chunk = True + if chunk.choices[0].delta.content[0] == "<": + accumulating_function_call = True + + if chunk.choices[0].delta.content: + if accumulating_function_call: + function_call_accumulator += chunk.choices[0].delta.content + else: + await self.push_frame(TextFrame(chunk.choices[0].delta.content)) + + if chunk.choices[0].finish_reason == 'eos' and accumulating_function_call: + await self._extract_function_call(context, function_call_accumulator) + + except CancelledError as e: + # todo: implement token counting estimates for use when the user interrupts a long generation + # we do this in the anthropic.py service + raise + except Exception as e: + logger.exception(f"{self} exception: {e}") + finally: + await self.stop_processing_metrics() + await self.push_frame(LLMFullResponseEndFrame()) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + context = None + if isinstance(frame, OpenAILLMContextFrame): + context = frame.context + elif isinstance(frame, LLMMessagesFrame): + context = TogetherLLMContext.from_messages(frame.messages) + elif isinstance(frame, LLMModelUpdateFrame): + logger.debug(f"Switching LLM model to: [{frame.model}]") + self._model = frame.model + else: + await self.push_frame(frame, direction) + + if context: + await self._process_context(context) + + async def _extract_function_call(self, context, function_call_accumulator): + context.add_message({"role": "assistant", "content": function_call_accumulator}) + + function_regex = r"(.*?)" + match = re.search(function_regex, function_call_accumulator) + if match: + function_name, args_string = match.groups() + try: + arguments = json.loads(args_string) + await self.call_function(context=context, + tool_call_id=uuid.uuid4(), + function_name=function_name, + arguments=arguments) + return + except json.JSONDecodeError as error: + # We get here if the LLM returns a function call with invalid JSON arguments. This could happen + # because of LLM non-determinism, or maybe more often because of user error in the prompt. + # Should we do anything more than log a warning? + logger.debug(f"Error parsing function arguments: {error}") + + +class TogetherLLMContext(OpenAILLMContext): + def __init__( + self, + messages: list[dict] | None = None, + ): + super().__init__(messages=messages) + + @classmethod + def from_openai_context(cls, openai_context: OpenAILLMContext): + self = cls( + messages=openai_context.messages, + ) + return self + + @classmethod + def from_messages(cls, messages: List[dict]) -> "TogetherLLMContext": + return cls(messages=messages) + + def add_message(self, message): + try: + self.messages.append(message) + except Exception as e: + logger.error(f"Error adding message: {e}") + + def get_messages_for_logging(self) -> str: + return json.dumps(self.messages) + + +class TogetherUserContextAggregator(LLMUserContextAggregator): + def __init__(self, context: OpenAILLMContext | TogetherLLMContext): + super().__init__(context=context) + + if isinstance(context, OpenAILLMContext): + self._context = TogetherLLMContext.from_openai_context(context) + + async def push_messages_frame(self): + frame = OpenAILLMContextFrame(self._context) + await self.push_frame(frame) + + 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] + except Exception as e: + logger.error(f"Error processing frame: {e}") + +# +# Claude returns a text content block along with a tool use content block. This works quite nicely +# with streaming. We get the text first, so we can start streaming it right away. Then we get the +# tool_use block. While the text is streaming to TTS and the transport, we can run the tool call. +# +# But Claude is verbose. It would be nice to come up with prompt language that suppresses Claude's +# chattiness about it's tool thinking. +# + + +class TogetherAssistantContextAggregator(LLMAssistantContextAggregator): + def __init__(self, user_context_aggregator: TogetherUserContextAggregator): + super().__init__(context=user_context_aggregator._context) + self._user_context_aggregator = user_context_aggregator + self._function_call_in_progress = None + self._function_call_result = None + + 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( + f"FunctionCallResultFrame tool_call_id does not match FunctionCallInProgressFrame tool_call_id") + self._function_call_in_progress = None + self._function_call_result = None + + def add_message(self, message): + self._user_context_aggregator.add_message(message) + + async def _push_aggregation(self): + if not (self._aggregation or self._function_call_result): + return + + run_llm = False + + aggregation = self._aggregation + self._aggregation = "" + + try: + if self._function_call_result: + frame = self._function_call_result + self._function_call_result = None + self._context.add_message({ + "role": "tool", + "content": frame.result + }) + run_llm = True + else: + self._context.add_message({"role": "assistant", "content": aggregation}) + + if run_llm: + await self._user_context_aggregator.push_messages_frame() + + except Exception as e: + logger.error(f"Error processing frame: {e}") diff --git a/src/pipecat/services/xtts.py b/src/pipecat/services/xtts.py index a4b144b9a..38f0f9a64 100644 --- a/src/pipecat/services/xtts.py +++ b/src/pipecat/services/xtts.py @@ -8,7 +8,13 @@ import aiohttp from typing import Any, AsyncGenerator, Dict -from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame, MetricsFrame, StartFrame +from pipecat.frames.frames import ( + AudioRawFrame, + ErrorFrame, + Frame, + StartFrame, + TTSStartedFrame, + TTSStoppedFrame) from pipecat.services.ai_services import TTSService from loguru import logger @@ -99,8 +105,9 @@ class XTTSService(TTSService): await self.start_tts_usage_metrics(text) - buffer = bytearray() + await self.push_frame(TTSStartedFrame()) + buffer = bytearray() async for chunk in r.content.iter_chunked(1024): if len(chunk) > 0: await self.stop_ttfb_metrics() @@ -131,3 +138,5 @@ class XTTSService(TTSService): resampled_audio_bytes = resampled_audio.astype(np.int16).tobytes() frame = AudioRawFrame(resampled_audio_bytes, 16000, 1) yield frame + + await self.push_frame(TTSStoppedFrame()) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index a4169991d..31cf2bff2 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -56,6 +56,11 @@ class BaseOutputTransport(FrameProcessor): self._stopped_event = asyncio.Event() + # Indicates if the bot is currently speaking. This is useful when we + # have an interruption since all the queued messages will be thrown + # away and we would lose the TTSStoppedFrame. + self._bot_speaking = False + # Create sink frame task. This is the task that will actually write # audio or video frames. We write audio/video in a task so we can keep # generating frames upstream while, for example, the audio is playing. @@ -151,6 +156,8 @@ class BaseOutputTransport(FrameProcessor): await self._handle_audio(frame) elif isinstance(frame, ImageRawFrame) or isinstance(frame, SpriteFrame): await self._handle_image(frame) + elif isinstance(frame, TransportMessageFrame) and frame.urgent: + await self.send_message(frame) else: await self._sink_queue.put(frame) @@ -167,6 +174,9 @@ class BaseOutputTransport(FrameProcessor): self._push_frame_task.cancel() await self._push_frame_task self._create_push_task() + # Let's send a bot stopped speaking if we have to. + if self._bot_speaking: + await self._bot_stopped_speaking() async def _handle_audio(self, frame: AudioRawFrame): if not self._params.audio_out_enabled: @@ -212,10 +222,10 @@ class BaseOutputTransport(FrameProcessor): elif isinstance(frame, TransportMessageFrame): await self.send_message(frame) elif isinstance(frame, TTSStartedFrame): - await self._internal_push_frame(BotStartedSpeakingFrame(), FrameDirection.UPSTREAM) + await self._bot_started_speaking() await self._internal_push_frame(frame) elif isinstance(frame, TTSStoppedFrame): - await self._internal_push_frame(BotStoppedSpeakingFrame(), FrameDirection.UPSTREAM) + await self._bot_stopped_speaking() await self._internal_push_frame(frame) else: await self._internal_push_frame(frame) @@ -228,6 +238,14 @@ class BaseOutputTransport(FrameProcessor): except Exception as e: logger.exception(f"{self} error processing sink queue: {e}") + async def _bot_started_speaking(self): + self._bot_speaking = True + await self._internal_push_frame(BotStartedSpeakingFrame(), FrameDirection.UPSTREAM) + + async def _bot_stopped_speaking(self): + self._bot_speaking = False + await self._internal_push_frame(BotStoppedSpeakingFrame(), FrameDirection.UPSTREAM) + # # Push frames task # diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 24a7f01bc..0d2510f1b 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -534,6 +534,7 @@ class DailyInputTransport(BaseInputTransport): self._client = client self._video_renderers = {} + self._audio_in_task = None self._vad_analyzer: VADAnalyzer | None = params.vad_analyzer if params.vad_enabled and not params.vad_analyzer: @@ -557,7 +558,7 @@ class DailyInputTransport(BaseInputTransport): # Leave the room. await self._client.leave() # Stop audio thread. - if self._params.audio_in_enabled or self._params.vad_enabled: + if self._audio_in_task and (self._params.audio_in_enabled or self._params.vad_enabled): self._audio_in_task.cancel() await self._audio_in_task @@ -567,7 +568,7 @@ class DailyInputTransport(BaseInputTransport): # Leave the room. await self._client.leave() # Stop audio thread. - if self._params.audio_in_enabled or self._params.vad_enabled: + if self._audio_in_task and (self._params.audio_in_enabled or self._params.vad_enabled): self._audio_in_task.cancel() await self._audio_in_task @@ -728,7 +729,7 @@ class DailyTransport(BaseTransport): room_url: str, token: str | None, bot_name: str, - params: DailyParams, + params: DailyParams = DailyParams(), input_name: str | None = None, output_name: str | None = None, loop: asyncio.AbstractEventLoop | None = None): @@ -793,7 +794,7 @@ class DailyTransport(BaseTransport): # DailyTransport # - @ property + @property def participant_id(self) -> str: return self._client.participant_id diff --git a/src/pipecat/transports/services/helpers/daily_rest.py b/src/pipecat/transports/services/helpers/daily_rest.py index 840222290..584cd7d67 100644 --- a/src/pipecat/transports/services/helpers/daily_rest.py +++ b/src/pipecat/transports/services/helpers/daily_rest.py @@ -70,9 +70,13 @@ class DailyRESTHelper: self.daily_api_url = daily_api_url self.aiohttp_session = aiohttp_session - def _get_name_from_url(self, room_url: str) -> str: + def get_name_from_url(self, room_url: str) -> str: return urlparse(room_url).path[1:] + async def get_room_from_url(self, room_url: str) -> DailyRoomObject: + room_name = self.get_name_from_url(room_url) + return await self._get_room_from_name(room_name) + async def create_room(self, params: DailyRoomParams) -> DailyRoomObject: headers = {"Authorization": f"Bearer {self.daily_api_key}"} json = {**params.model_dump(exclude_none=True)} @@ -90,25 +94,6 @@ class DailyRESTHelper: return room - async def _get_room_from_name(self, room_name: str) -> DailyRoomObject: - headers = {"Authorization": f"Bearer {self.daily_api_key}"} - async with self.aiohttp_session.get(f"{self.daily_api_url}/rooms/{room_name}", headers=headers) as r: - if r.status != 200: - raise Exception(f"Room not found: {room_name}") - - data = await r.json() - - try: - room = DailyRoomObject(**data) - except ValidationError as e: - raise Exception(f"Invalid response: {e}") - - return room - - async def get_room_from_url(self, room_url: str,) -> DailyRoomObject: - room_name = self._get_name_from_url(room_url) - return await self._get_room_from_name(room_name) - async def get_token( self, room_url: str, @@ -120,7 +105,7 @@ class DailyRESTHelper: expiration: float = time.time() + expiry_time - room_name = self._get_name_from_url(room_url) + room_name = self.get_name_from_url(room_url) headers = {"Authorization": f"Bearer {self.daily_api_key}"} json = { @@ -139,12 +124,29 @@ class DailyRESTHelper: return data["token"] + async def delete_room_by_url(self, room_url: str) -> bool: + room_name = self.get_name_from_url(room_url) + return await self.delete_room_by_name(room_name) + async def delete_room_by_name(self, room_name: str) -> bool: headers = {"Authorization": f"Bearer {self.daily_api_key}"} async with self.aiohttp_session.delete(f"{self.daily_api_url}/rooms/{room_name}", headers=headers) as r: if r.status != 200 and r.status != 404: raise Exception(f"Failed to delete room: {room_name}") + return True + + async def _get_room_from_name(self, room_name: str) -> DailyRoomObject: + headers = {"Authorization": f"Bearer {self.daily_api_key}"} + async with self.aiohttp_session.get(f"{self.daily_api_url}/rooms/{room_name}", headers=headers) as r: + if r.status != 200: + raise Exception(f"Room not found: {room_name}") + data = await r.json() - return True + try: + room = DailyRoomObject(**data) + except ValidationError as e: + raise Exception(f"Invalid response: {e}") + + return room diff --git a/src/pipecat/utils/string.py b/src/pipecat/utils/string.py new file mode 100644 index 000000000..a47db6c5c --- /dev/null +++ b/src/pipecat/utils/string.py @@ -0,0 +1,24 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import re + + +ENDOFSENTENCE_PATTERN_STR = r""" + (? bool: + return ENDOFSENTENCE_PATTERN.search(text.rstrip()) is not None