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

Merge pull request #352 from pipecat-ai/aleix/rtvi-0.1

Browse Source 
processors(rtvi): rtvi 0.1 message protocol
This commit is contained in:
Aleix Conchillo Flaqué
2024-08-15 17:35:50 -07:00
committed by GitHub
parent f0e50ba000 4a6638f749
commit b57c9d569b
29 changed files with 2011 additions and 607 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
examples/foundational/12c-describe-video-anthropic.py
View File

@@ -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")

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

@@ -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)

122
examples/foundational/19a-tools-anthropic.py Normal file
View File

@@ -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())

183
examples/foundational/19b-tools-video-anthropic.py Normal file
View File

@@ -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())

137
examples/foundational/19c-tools-togetherai.py Normal file
View File

@@ -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:
<function=example_function_name>{{\"example_name\": \"example_value\"}}</function>
Reminder:
- Function calls MUST follow the specified format, start with <function= and end with </function>
- 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())

2
linux-py3.10-requirements.txt
View File

@@ -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)

2
macos-py3.10-requirements.txt
View File

@@ -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)

3
pyproject.toml
View File

@@ -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" ]

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

@@ -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

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

@@ -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

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

@@ -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:

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

@@ -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

13
src/pipecat/processors/aggregators/sentence.py
View File

@@ -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))

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

@@ -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}"

23
src/pipecat/processors/logger.py
View File

@@ -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)

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

@@ -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"""
(?<![A-Z]) # Negative lookbehind: not preceded by an uppercase letter (e.g., "U.S.A.")
(?<!\d) # Negative lookbehind: not preceded by a digit (e.g., "1. Let's start")
(?<!\d\s[ap]) # Negative lookbehind: not preceded by time (e.g., "3:00 a.m.")
(?<!Mr|Ms|Dr) # Negative lookbehind: not preceded by Mr, Ms, Dr (combined bc. length is the same)
(?<!Mrs) # Negative lookbehind: not preceded by "Mrs"
(?<!Prof) # Negative lookbehind: not preceded by "Prof"
[\.\?\!:] # Match a period, question mark, exclamation point, or colon
$ # End of string
"""
ENDOFSENTENCE_PATTERN = re.compile(ENDOFSENTENCE_PATTERN_STR, re.VERBOSE)
def match_endofsentence(text: str) -> 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.

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

@@ -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}")

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

@@ -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}")

16
src/pipecat/services/cartesia.py
View File

@@ -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

5
src/pipecat/services/deepgram.py
View File

@@ -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}")

4
src/pipecat/services/elevenlabs.py
View File

@@ -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())

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

@@ -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}")

4
src/pipecat/services/playht.py
View File

@@ -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}")

314
src/pipecat/services/together.py Normal file
View File

@@ -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"<function=(\w+)>(.*?)</function>"
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}")

13
src/pipecat/services/xtts.py
View File

@@ -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())

22
src/pipecat/transports/base_output.py
View File

@@ -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
#

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

@@ -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

46
src/pipecat/transports/services/helpers/daily_rest.py
View File

@@ -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

24
src/pipecat/utils/string.py Normal file
View File

@@ -0,0 +1,24 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import re
ENDOFSENTENCE_PATTERN_STR = r"""
(?<![A-Z]) # Negative lookbehind: not preceded by an uppercase letter (e.g., "U.S.A.")
(?<!\d) # Negative lookbehind: not preceded by a digit (e.g., "1. Let's start")
(?<!\d\s[ap]) # Negative lookbehind: not preceded by time (e.g., "3:00 a.m.")
(?<!Mr|Ms|Dr) # Negative lookbehind: not preceded by Mr, Ms, Dr (combined bc. length is the same)
(?<!Mrs) # Negative lookbehind: not preceded by "Mrs"
(?<!Prof) # Negative lookbehind: not preceded by "Prof"
[\.\?\!:] # Match a period, question mark, exclamation point, or colon
$ # End of string
"""
ENDOFSENTENCE_PATTERN = re.compile(ENDOFSENTENCE_PATTERN_STR, re.VERBOSE)
def match_endofsentence(text: str) -> bool:
return ENDOFSENTENCE_PATTERN.search(text.rstrip()) is not None